অনুশীলনে স্পার্ক স্কিমা বিবর্তন

প্রিয় পাঠক, শুভ বিকাল!

এই নিবন্ধে, নেওফ্লেক্সের বিগ ডেটা সলিউশন ব্যবসায়িক এলাকার শীর্ষস্থানীয় পরামর্শদাতা অ্যাপাচি স্পার্ক ব্যবহার করে পরিবর্তনশীল কাঠামো স্টোরফ্রন্ট তৈরির জন্য বিশদ বিকল্পগুলি বর্ণনা করেছেন।

একটি ডেটা বিশ্লেষণ প্রকল্পের অংশ হিসাবে, ঢিলেঢালাভাবে স্ট্রাকচার্ড ডেটার উপর ভিত্তি করে শোকেস তৈরি করার কাজটি প্রায়ই দেখা দেয়।

সাধারণত এগুলি হল লগ, বা বিভিন্ন সিস্টেমের প্রতিক্রিয়া, JSON বা XML আকারে সংরক্ষিত। ডেটা Hadoop এ আপলোড করা হয়, তারপর এটি থেকে একটি স্টোরফ্রন্ট তৈরি করা প্রয়োজন। আমরা তৈরি স্টোরফ্রন্টে অ্যাক্সেস সংগঠিত করতে পারি, উদাহরণস্বরূপ, ইম্পালার মাধ্যমে।

এই ক্ষেত্রে, লক্ষ্য স্টোরফ্রন্টের লেআউট আগাম অজানা। তদুপরি, স্কিমটি আগে থেকে তৈরি করা যাবে না, যেহেতু এটি ডেটার উপর নির্ভর করে এবং আমরা এই খুব দুর্বল কাঠামোগত ডেটা নিয়ে কাজ করছি।

উদাহরণস্বরূপ, আজ নিম্নলিখিত প্রতিক্রিয়াটি লগ করা হয়েছে:

{source: "app1", error_code: ""}

এবং আগামীকাল নিম্নলিখিত প্রতিক্রিয়া একই সিস্টেম থেকে আসে:

{source: "app1", error_code: "error", description: "Network error"}

ফলস্বরূপ, স্টোরফ্রন্টে আরেকটি ক্ষেত্র যোগ করা উচিত - বিবরণ, এবং কেউ জানে না এটি আসবে কি না।

এই জাতীয় ডেটাতে একটি মার্ট তৈরি করার কাজটি মোটামুটি মানক, এবং এর জন্য স্পার্কের বেশ কয়েকটি সরঞ্জাম রয়েছে। উত্স ডেটা পার্স করার জন্য, JSON এবং XML উভয়ের জন্যই সমর্থন রয়েছে এবং পূর্বে অজানা স্কিমার জন্য, স্কিমাইভোলিউশন সমর্থন প্রদান করা হয়।

প্রথম নজরে, সমাধান সহজ দেখায়। আপনাকে JSON এর সাথে ফোল্ডারটি নিতে হবে এবং ডেটাফ্রেমে পড়তে হবে। স্পার্ক একটি স্কিমা তৈরি করবে এবং নেস্টেড ডেটাকে স্ট্রাকচারে পরিণত করবে। এর পরে, হাইভ মেটাস্টোরে স্টোরফ্রন্ট নিবন্ধন করার মাধ্যমে, সমস্ত কিছু parquet-এ সংরক্ষণ করা প্রয়োজন, যা ইমপালাতেও সমর্থিত।

সবকিছু সহজ বলে মনে হচ্ছে।

যাইহোক, ডকুমেন্টেশনের সংক্ষিপ্ত উদাহরণগুলি থেকে অনুশীলনে বেশ কয়েকটি সমস্যা নিয়ে কী করা উচিত তা স্পষ্ট নয়।

ডকুমেন্টেশন একটি স্টোরফ্রন্ট তৈরি করার জন্য নয়, কিন্তু একটি ডেটাফ্রেমে JSON বা XML পড়ার জন্য একটি পদ্ধতির বর্ণনা করে।

যথা, এটি সহজভাবে দেখায় কিভাবে JSON পড়তে এবং পার্স করতে হয়:

df = spark.read.json(path...)

এটি স্পার্কের কাছে ডেটা উপলব্ধ করার জন্য যথেষ্ট।

বাস্তবে, দৃশ্যকল্পটি কেবল একটি ফোল্ডার থেকে JSON ফাইলগুলি পড়া এবং একটি ডেটাফ্রেম তৈরি করার চেয়ে অনেক বেশি জটিল। পরিস্থিতি এইরকম দেখাচ্ছে: ইতিমধ্যে একটি নির্দিষ্ট শোকেস রয়েছে, প্রতিদিন নতুন ডেটা আসে, সেগুলি শোকেসে যুক্ত করা দরকার, ভুলে যাবেন না যে স্কিমটি আলাদা হতে পারে।

একটি স্টোরফ্রন্ট নির্মাণের জন্য সাধারণ স্কিম নিম্নরূপ:

1 পর্যায়। ডেটা Hadoop-এ লোড করা হয়, তারপরে দৈনিক অতিরিক্ত লোড করা হয় এবং একটি নতুন পার্টিশনে যোগ করা হয়। ফলাফলটি উত্স ডেটা সহ একটি ফোল্ডার, যা দিনে বিভাজিত হয়।

2 পর্যায়। প্রাথমিক লোডিংয়ের সময়, এই ফোল্ডারটি স্পার্ক ব্যবহার করে পড়া এবং পার্স করা হয়। ফলস্বরূপ ডেটাফ্রেমটি এমন একটি বিন্যাসে সংরক্ষিত হয় যা বিশ্লেষণ করা যেতে পারে, উদাহরণস্বরূপ, parquet এ, যা পরে ইমপালাতে আমদানি করা যেতে পারে। এটি এই বিন্দু পর্যন্ত জমা হওয়া সমস্ত ডেটা সহ একটি লক্ষ্য স্টোরফ্রন্ট তৈরি করে।

3 পর্যায়। একটি ডাউনলোড তৈরি করা হয়েছে যা প্রতিদিন স্টোরফ্রন্ট আপডেট করবে।
ক্রমবর্ধমান লোডিংয়ের প্রশ্ন ওঠে, স্টোরফ্রন্টকে পার্টিশন করার প্রয়োজন এবং স্টোরফ্রন্টের সাধারণ বিন্যাসকে সমর্থন করার প্রশ্ন।

একটা উদাহরণ দেওয়া যাক। ধরা যাক একটি সংগ্রহস্থল তৈরির প্রথম ধাপটি বাস্তবায়িত হয়েছে এবং একটি ফোল্ডারে JSON ফাইল আপলোড করা কনফিগার করা হয়েছে৷

তাদের থেকে একটি ডেটাফ্রেম তৈরি করতে এবং তারপরে একটি শোকেস হিসাবে সংরক্ষণ করতে কোনও সমস্যা নেই। এটি হল প্রথম ধাপ যা সহজেই স্পার্ক ডকুমেন্টেশনে পাওয়া যাবে:

df = spark.read.option("mergeSchema", True).json(".../*") 
df.printSchema()

root 
|-- a: long (nullable = true) 
|-- b: string (nullable = true) 
|-- c: struct (nullable = true) |    
|-- d: long (nullable = true)

সবকিছু ঠিক হবে বলে মনে হয়।

আমরা JSON পড়েছি এবং পার্স করেছি, তারপরে আমরা ডাটাফ্রেমটিকে parquet হিসাবে সংরক্ষণ করি, যেকোনো সুবিধাজনক উপায়ে Hive-এ নিবন্ধন করি:

df.write.format(“parquet”).option('path','<External Table Path>').saveAsTable('<Table Name>')

আমরা একটি শোকেস পেতে.

কিন্তু, পরের দিন উৎস থেকে নতুন তথ্য যোগ করা হয়। আমাদের কাছে JSON এর সাথে একটি ফোল্ডার রয়েছে এবং এই ফোল্ডারের উপর ভিত্তি করে একটি স্টোরফ্রন্ট তৈরি করা হয়েছে। উৎস থেকে ডেটার পরবর্তী অংশ লোড করার পর, স্টোরফ্রন্টে একদিনের জন্য পর্যাপ্ত ডেটা থাকে না।

একটি যৌক্তিক সমাধান হ'ল স্টোরফ্রন্টকে দিনে বিভাজন করা, যা আপনাকে পরের দিন একটি নতুন পার্টিশন যুক্ত করার অনুমতি দেবে। এটির প্রক্রিয়াটিও সুপরিচিত; স্পার্ক আপনাকে আলাদাভাবে পার্টিশন রেকর্ড করতে দেয়।

প্রথমত, আমরা প্রাথমিক লোডিং করি, উপরে বর্ণিত ডেটা সংরক্ষণ করি, শুধুমাত্র পার্টিশন যোগ করি। এই ক্রিয়াটিকে স্টোরফ্রন্ট ইনিশিয়ালাইজেশন বলা হয় এবং এটি শুধুমাত্র একবার করা হয়:

df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)

পরের দিন আমরা শুধুমাত্র নতুন পার্টিশন ডাউনলোড করব:

df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")

স্কিমা আপডেট করতে Hive-এ পুনঃনিবন্ধন করা বাকি আছে।
যাইহোক, এখানেই সমস্যা দেখা দেয়।

প্রথম সমস্যা। শীঘ্রই বা পরে, ফলস্বরূপ কাঠি আর পাঠযোগ্য হবে না। এই কারণে যেভাবে parquet এবং JSON খালি ক্ষেত্রগুলিকে ভিন্নভাবে আচরণ করে।

আসুন একটি সাধারণ পরিস্থিতি বিবেচনা করা যাক। উদাহরণস্বরূপ, গতকাল JSON এসেছে:

День 1: {"a": {"b": 1}},

এবং আজ একই JSON এর মত দেখাচ্ছে:

День 2: {"a": null}

ধরা যাক আমাদের দুটি আলাদা পার্টিশন আছে, প্রতিটিতে একটি লাইন আছে।
যখন আমরা পুরো উৎসের ডেটা পড়ি, তখন স্পার্ক টাইপ নির্ধারণ করতে সক্ষম হবে, এবং বুঝতে পারবে যে "a" হল "কাঠামো" টাইপের একটি ক্ষেত্র, যার একটি নেস্টেড ক্ষেত্র "b" টাইপ INT। তবে, যদি প্রতিটি পার্টিশন আলাদাভাবে সংরক্ষণ করা হয়, তবে ফলাফলটি বেমানান পার্টিশন স্কিমগুলির সাথে একটি কাঠের তৈরি:

df1 (a: <struct<"b": INT>>)
df2 (a: STRING NULLABLE)

এই পরিস্থিতিটি সুপরিচিত, তাই উৎস ডেটা পার্স করার সময় খালি ক্ষেত্রগুলি সরানোর জন্য একটি বিকল্প বিশেষভাবে যোগ করা হয়েছে:

df = spark.read.json("...", dropFieldIfAllNull=True)

এই ক্ষেত্রে, parquet একসঙ্গে পড়া যাবে যে পার্টিশন গঠিত হবে।
যদিও অনুশীলনে যারা এটি করেছেন তারা তিক্তভাবে হাসবেন। কেন? হ্যাঁ, কারণ সম্ভবত আরও দুটি পরিস্থিতি দেখা দেবে। অথবা তিন. অথবা চার. প্রথমটি, যা প্রায় নিশ্চিত, বিভিন্ন JSON ফাইলে সাংখ্যিক প্রকারগুলি ভিন্ন দেখাবে৷ উদাহরণস্বরূপ, {intField: 1} এবং {intField: 1.1}। যদি এই জাতীয় ক্ষেত্রগুলি একটি ব্যাচে উপস্থিত হয়, তবে স্কিমা মার্জটি সঠিকভাবে সবকিছু পড়বে, যা সবচেয়ে সঠিক টাইপের দিকে নিয়ে যাবে। কিন্তু বিভিন্ন বেশী হলে, তারপর একটি intField থাকবে: int, এবং অন্য intField থাকবে: ডবল।

এই পরিস্থিতি পরিচালনা করার জন্য নিম্নলিখিত পতাকা আছে:

df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)

এখন আমাদের কাছে একটি ফোল্ডার রয়েছে যেখানে পার্টিশনগুলি অবস্থিত, যা একটি একক ডেটাফ্রেমে এবং পুরো স্টোরফ্রন্টের একটি বৈধ প্যারকেট পড়তে পারে। হ্যাঁ? না.

আমাদের মনে রাখতে হবে যে আমরা Hive এ টেবিল নিবন্ধন করেছি। Hive ক্ষেত্রের নামের ক্ষেত্রে কেস সংবেদনশীল নয়, কিন্তু parquet হয়। অতএব, স্কিমা সহ পার্টিশন: field1: int এবং Field1: int হাইভের জন্য একই, কিন্তু স্পার্কের জন্য নয়। ছোট হাতের ক্ষেত্রে ক্ষেত্রের নাম পরিবর্তন করতে ভুলবেন না।

এর পরে, সবকিছু ঠিক আছে বলে মনে হচ্ছে।

যাইহোক, সব এত সহজ নয়। একটি দ্বিতীয়, এছাড়াও সুপরিচিত সমস্যা দেখা দেয়. যেহেতু প্রতিটি নতুন পার্টিশন আলাদাভাবে সংরক্ষিত হয়, পার্টিশন ফোল্ডারে স্পার্ক সার্ভিস ফাইল থাকবে, উদাহরণস্বরূপ, _SUCCESS অপারেশন সফল পতাকা। কাঠবাদাম করার চেষ্টা করার সময় এটি একটি ত্রুটির কারণ হবে। এটি এড়ানোর জন্য, আপনাকে ফোল্ডারে পরিষেবা ফাইলগুলি যোগ করা থেকে স্পার্ককে প্রতিরোধ করে কনফিগারেশনটি কনফিগার করতে হবে:

hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

মনে হচ্ছে এখন প্রতিদিন একটি নতুন প্যারকেট পার্টিশন টার্গেট স্টোরফ্রন্ট ফোল্ডারে যোগ করা হয়, যেখানে দিনের পার্স করা ডেটা থাকে। ডাটা টাইপ দ্বন্দ্ব সহ কোনও পার্টিশন যাতে না থাকে তা নিশ্চিত করার জন্য আমরা আগে থেকেই যত্ন নিয়েছিলাম।

কিন্তু আমরা তৃতীয় সমস্যার সম্মুখীন হয়েছি। এখন সাধারণ স্কিমা জানা নেই, তাছাড়া, Hive-এ টেবিলের ভুল স্কিমা রয়েছে, যেহেতু প্রতিটি নতুন পার্টিশন সম্ভবত স্কিমাতে একটি বিকৃতি প্রবর্তন করে।

টেবিল পুনরায় নিবন্ধন করা প্রয়োজন. এটি সহজভাবে করা যেতে পারে: স্টোরফ্রন্টের প্যারকেটটি আবার পড়ুন, স্কিমা নিন এবং এটির উপর ভিত্তি করে একটি DDL তৈরি করুন, যার সাহায্যে আপনি Hive-এ ফোল্ডারটিকে একটি বাহ্যিক টেবিল হিসাবে পুনরায় নিবন্ধন করতে পারেন, লক্ষ্য স্টোরফ্রন্টের স্কিমা আপডেট করতে পারেন৷

আমরা একটি চতুর্থ সমস্যা সম্মুখীন. যখন আমরা প্রথমবার টেবিল নিবন্ধন করি, তখন আমরা স্পার্কের উপর নির্ভর করেছিলাম। এখন আমরা এটি নিজেরাই করি, এবং আমাদের মনে রাখা দরকার যে কাঠের ক্ষেত্রগুলি হাইভ দ্বারা অনুমোদিত নয় এমন অক্ষর দিয়ে শুরু হতে পারে। উদাহরণস্বরূপ, স্পার্ক লাইনগুলি ছুঁড়ে দেয় যে এটি "corrupt_record" ক্ষেত্রে পার্স করতে পারে না। এই ধরনের একটি ক্ষেত্র পলায়ন ছাড়া Hive নিবন্ধিত করা যাবে না.

এটি জেনে, আমরা চিত্রটি পাই:

f_def = ""
for f in pf.dtypes:
  if f[0] != "date_load":
    f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") 
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '/user/admin/testJson/testSchemaEvolution/pq/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)

কোড ("_corrupt_record", "`_corrupt_record`") + " " + f[1].প্রতিস্থাপন(":", "`:").প্রতিস্থাপন("<", "<`").প্রতিস্থাপন(",", ",`").প্রতিস্থাপন("অ্যারে<`", "অ্যারে<") নিরাপদ DDL করে, অর্থাৎ এর পরিবর্তে:

create table tname (_field1 string, 1field string)

“_field1, 1field” এর মত ক্ষেত্রের নাম দিয়ে, একটি নিরাপদ DDL তৈরি করা হয় যেখানে ক্ষেত্রের নামগুলি এস্কেপ করা হয়: টেবিল `tname` (`_field1` স্ট্রিং, `1field` স্ট্রিং) তৈরি করুন।

প্রশ্ন উঠছে: কীভাবে সঠিকভাবে একটি সম্পূর্ণ স্কিমা (পিএফ কোডে) সহ একটি ডেটাফ্রেম পাবেন? কিভাবে এই পিএফ পেতে? এটি পঞ্চম সমস্যা। টার্গেট স্টোরফ্রন্টের parquet ফাইল সহ ফোল্ডার থেকে সমস্ত পার্টিশনের চিত্রটি পুনরায় পড়ুন? এই পদ্ধতিটি সবচেয়ে নিরাপদ, কিন্তু কঠিন।

স্কিমা ইতিমধ্যে Hive মধ্যে আছে. আপনি পুরো টেবিলের স্কিমা এবং নতুন পার্টিশন একত্রিত করে একটি নতুন স্কিমা পেতে পারেন। এর মানে হল যে আপনাকে Hive থেকে টেবিল স্কিমা নিতে হবে এবং এটিকে নতুন পার্টিশনের স্কিমার সাথে একত্রিত করতে হবে। এটি হাইভ থেকে পরীক্ষার মেটাডেটা পড়ার মাধ্যমে, একটি অস্থায়ী ফোল্ডারে সংরক্ষণ করে এবং স্পার্ক ব্যবহার করে উভয় পার্টিশন একবারে পড়ার মাধ্যমে করা যেতে পারে।

মূলত, আপনার যা যা প্রয়োজন তা রয়েছে: Hive-এ মূল টেবিল স্কিমা এবং একটি নতুন পার্টিশন। আমাদের কাছে ডেটাও আছে। যা অবশিষ্ট থাকে তা হল একটি নতুন স্কিমা যা তৈরি করা পার্টিশন থেকে স্টোরফ্রন্ট স্কিমা এবং নতুন ক্ষেত্রগুলিকে একত্রিত করে:

from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")

এর পরে, আমরা পূর্ববর্তী খণ্ডের মতো টেবিল নিবন্ধন DDL তৈরি করি।
যদি পুরো চেইনটি সঠিকভাবে কাজ করে, যথা, একটি প্রাথমিক লোড ছিল এবং টেবিলটি হাইভে সঠিকভাবে তৈরি করা হয়েছিল, তাহলে আমরা একটি আপডেট করা টেবিল স্কিমা পাই।

শেষ সমস্যা হল যে আপনি সহজে একটি Hive টেবিলে একটি পার্টিশন যোগ করতে পারবেন না, কারণ এটি ভেঙে যাবে। আপনাকে Hive এর পার্টিশন কাঠামো ঠিক করতে বাধ্য করতে হবে:

from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)

JSON পড়ার এবং এর উপর ভিত্তি করে একটি স্টোরফ্রন্ট তৈরি করার সহজ কাজটি বেশ কয়েকটি অন্তর্নিহিত অসুবিধা কাটিয়ে উঠতে পারে, যার সমাধানগুলি অবশ্যই আলাদাভাবে খুঁজে বের করতে হবে। এবং যদিও এই সমাধানগুলি সহজ, তবে সেগুলি খুঁজে পেতে অনেক সময় লাগে।

একটি শোকেস নির্মাণ বাস্তবায়ন করতে, আমাদের ছিল:

• স্টোরফ্রন্টে পার্টিশন যোগ করুন, পরিষেবা ফাইলগুলি থেকে মুক্তি পান
• স্পার্ক টাইপ করেছে এমন উত্স ডেটাতে খালি ক্ষেত্রগুলির সাথে ডিল করুন৷
• স্ট্রিং-এ সহজ প্রকার কাস্ট করুন
• ক্ষেত্রের নাম ছোট হাতের অক্ষরে রূপান্তর করুন
• Hive-এ পৃথক ডেটা আপলোড এবং টেবিল নিবন্ধন (DDL সৃষ্টি)
• Hive এর সাথে সামঞ্জস্যপূর্ণ নাও হতে পারে এমন ক্ষেত্রের নাম এড়িয়ে যেতে ভুলবেন না
• Hive-এ টেবিল রেজিস্ট্রেশন আপডেট করতে শিখুন

সংক্ষেপে, আমরা নোট করি যে স্টোরফ্রন্টগুলি তৈরি করার সিদ্ধান্তটি অনেকগুলি ত্রুটিতে পরিপূর্ণ। অতএব, যদি বাস্তবায়নে অসুবিধা দেখা দেয় তবে সফল দক্ষতা সহ একজন অভিজ্ঞ অংশীদারের কাছে যাওয়া ভাল।

এই নিবন্ধটি পড়ার জন্য আপনাকে ধন্যবাদ, আমরা আশা করি আপনি তথ্যটি দরকারী বলে মনে করেন।

উত্স: www.habr.com