実際のSparkschemaEvolution

読者の皆様、良い一日を！



この記事では、NeoflexのBig Data Solutionsビジネスラインの主要コンサルタントが、ApacheSparkを使用して可変構造ストアフロントを構築するためのオプションについて詳しく説明します。



データ分析プロジェクトの一環として、大まかに構造化されたデータに基づいてマートを構築するタスクが頻繁に発生します。



通常、これらはログ、またはさまざまなシステムからの応答であり、JSONまたはXMLとして保存されます。データはHadoopにアップロードされ、そこからショーケースを作成する必要があります。たとえば、Impalaを介して、作成されたストアフロントへのアクセスを整理できます。



この場合、ターゲットストアフロントのレイアウトは以前は不明でした。また、データに依存するため、事前に図を作成することはできず、これらの非常に弱い構造のデータを扱っています。



たとえば、今日、次の回答がログに記録されます。

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

そして明日、次の答えは同じシステムから来ます：

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

その結果、ストアフロントにもう1つのフィールド（説明）を追加する必要があり、それが来るかどうかは誰にもわかりません。



このようなデータでマートを作成するタスクはかなり標準的であり、Sparkにはこのためのツールがいくつかあります。生データの解析にはJSONとXMLの両方がサポートされており、schemaEvolutionのサポートはこれまで知られていなかったスキーマに提供されます。



一見、ソリューションは単純に見えます。JSONを含むフォルダーを取得し、それをデータフレームに読み込む必要があります。Sparkはスキーマを作成し、ネストされたデータを構造に変換します。次に、ショーケースをHiveメタストアに登録することにより、すべてを寄木細工の床に保存する必要があります。これはImpalaでもサポートされています。



すべてが単純なようです。



ただし、ドキュメントの短い例から、実際に多くの問題をどう処理するかは明確ではありません。



ドキュメントでは、ストアフロントを作成するためではなく、JSONまたはXMLをデータフレームに読み込むためのアプローチについて説明しています。



つまり、JSONの読み取りと解析の方法が簡単に示されています。

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

これは、Sparkがデータを利用できるようにするのに十分です。



実際には、シナリオは、フォルダからJSONファイルを読み取ってデータフレームを作成するよりもはるかに複雑です。状況は次のようになります。すでに特定のショーケースがあり、新しいデータが毎日届きます。スキームが異なる可能性があることを忘れずに、ショーケースに追加する必要があります。



ストアフロントを構築するための通常のスキームは次のとおりです。



ステップ1.データがHadoopにロードされ、その後毎日リロードされて新しいパーティションに追加されます。初期データが日数で分割されたフォルダが見つかります。



ステップ2。起動の初期化中に、このフォルダーはSparkによって読み取られて解析されます。結果のデータフレームは、たとえば寄木細工などの分析に使用できる形式で保存され、Impalaにインポートできます。これにより、これまでに蓄積されたすべてのデータを含むターゲットショーケースが作成されます。



ステップ3.ストアフロントを毎日更新するダウンロードが作成されます。

問題は、増分ロード、ショーケースを分割する必要性、およびショーケースの一般的なスキームをサポートすることの問題から生じます。



例を挙げましょう。ストレージを構築する最初のステップが実装され、JSONファイルのフォルダーへのエクスポートが構成されているとします。



それらからデータフレームを作成し、それをショーケースとして保存することは問題ではありません。これは、Sparkのドキュメントで簡単に見つけることができる最初のステップです。

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を読み取って解析し、データフレームを寄木細工として保存し、便利な方法でHiveに登録します。

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

ショーケースがあります。



しかし、翌日、ソースからの新しいデータが追加されました。JSONを含むフォルダーと、このフォルダーに基づいて作成されたショーケースがあります。ソースからデータの次のチャンクをロードした後、データマートは1日データを使い果たします。



論理的な解決策は、ストアフロントを日ごとにパーティション化することです。これにより、翌日ごとに新しいパーティションを追加できます。このメカニズムもよく知られています。Sparkを使用すると、パーティションを個別に書き込むことができます。



まず、初期化の読み込みを行い、上記のようにデータを保存し、パーティショニングのみを追加します。このアクションはストアフロント初期化と呼ばれ、1回だけ実行されます。

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に再登録することだけです。

ただし、ここで問題が発生します。



最初の問題。遅かれ早かれ、結果の寄木細工は読むことができなくなります。これは、寄木細工とJSONが空のフィールドにどのようにアプローチするかと関係があります。



典型的な状況を考えてみましょう。たとえば、JSONは昨日到着します。

 1: {"a": {"b": 1}},

そして今日、同じJSONは次のようになります。

 2: {"a": null}

それぞれ1行の2つの異なるパーティションがあるとしましょう。

生データ全体を読み取ると、Sparkはタイプを判別し、「a」がタイプ「structure」のフィールドであり、タイプINTのネストされたフィールド「b」があることを理解できます。ただし、各パーティションが個別に保存された場合、互換性のないパーティションスキームを持つ寄木細工が取得されます。

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

この状況はよく知られているため、初期データを解析するときに空のフィールドを削除するオプションが特別に追加されました。

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

この場合、寄木細工は一緒に読み取ることができるパーティションで構成されます。

実際にこれをした人は激しく笑うでしょうが。どうして？さらに2つの状況が発生する可能性があるためです。または3つ。または4つ。最初のものは、ほぼ確実に表示されますが、数値タイプはJSONファイルごとに異なって見えるということです。たとえば、{intField：1}および{intField：1.1}。そのようなフィールドが1つの部分で見つかった場合、スキーママージはすべてを正しく読み取り、最も正確なタイプになります。ただし、異なる場合、一方はintField：intになり、もう一方はintField：doubleになります。



この状況を処理するための次のフラグがあります。

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

これで、パーティションが配置されているフォルダーができました。このフォルダーは、単一のデータフレームに読み込むことができ、ストアフロント全体に有効な寄木細工です。はい？番号。



テーブルをHiveに登録したことを思い出してください。ハイブはフィールド名で大文字と小文字を区別しませんが、寄木細工は大文字と小文字を区別します。したがって、スキーマがfield1：intおよびField1：intのパーティションは、Hiveでは同じですが、Sparkでは同じではありません。フィールド名は小文字にすることを忘れないでください。



その後、すべてが順調に見えます。



ただし、すべてがそれほど単純というわけではありません。次に、よく知られている問題が発生します。新しいパーティションはそれぞれ個別に保存されるため、Sparkサービスファイルはパーティションフォルダにあります（例：_SUCCESS操作成功フラグ）。これは、寄木細工をしようとしたときにエラーをスローします。これを回避するには、Sparkがフォルダにサービスファイルを追加できないようにして構成を設定する必要があります。

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

現在、毎日新しい寄木細工のパーティションがターゲットのストアフロントフォルダーに追加され、その日の解析済みデータが保存されているようです。データタイプが競合するパーティションがないように事前に注意しました。



しかし、私たちの前に3番目の問題があります。さらに、Hiveでは、新しいパーティションごとにスキームに歪みが生じた可能性が高いため、スキームが間違っているテーブルの一般的なスキームは不明です。



テーブルを再登録する必要があります。これは簡単に行うことができます。ストアフロント寄木細工をもう一度読み、スキーマを取得し、それに基づいてDDLを作成します。これにより、Hiveのフォルダーを外部テーブルとして再登録し、ターゲットストアフロントスキーマを更新します。



私たちは4番目の問題に直面しています。初めてテーブルを登録したときは、Sparkに依存していました。今、私たちは自分たちでそれを行います、そしてあなたは寄木細工のフィールドがハイブに無効な文字で始まることができることを覚えておく必要があります。たとえば、Sparkは、解析できなかった行を「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] .replace（"： "、" `："）。Replace（ "<"、 "<` "）。Replace（"、 " 、 "、` "）。replace（" array <`"、 "array <"）は、DDLを安全にします。つまり、次の代わりになります。

create table tname (_field1 string, 1field string)

「_field1、1field」などのフィールド名を使用すると、フィールド名がエスケープされた場所で安全なDDLが作成されます。createtable `tname`（` _field1` string、 `1field` string）。



疑問が生じます：完全なスキーマでデータフレームを正しく（pfコードで）取得する方法は？このpfを取得するにはどうすればよいですか？これは5番目の問題です。ターゲットストアフロントの寄木細工のファイルを含むフォルダーからすべてのパーティションのスキーマを再読み込みしますか？これは最も安全な方法ですが、最も難しい方法です。



スキーマはすでにHiveにあります。テーブル全体のスキーマと新しいパーティションを組み合わせることで、新しいスキーマを取得できます。したがって、Hiveからテーブルスキーマを取得し、それを新しいパーティションスキーマと組み合わせる必要があります。これは、Hiveからテストメタデータを読み取り、それを一時フォルダに保存し、Sparkで両方のパーティションを一度に読み取ることで実行できます。



基本的に、必要なものはすべて揃っています。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テーブルにパーティションを追加するだけでは不十分なことです。Hiveにパーティション構造を修正させる必要があります。

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

JSONを読み取り、それに基づいてストアフロントを作成するという単純なタスクは、多くの暗黙の問題を克服することにつながります。その解決策は個別に探す必要があります。これらの解決策は単純ですが、見つけるのに長い時間がかかります。



ショーケースの建設を実施するために、私はしなければなりませんでした：

• ストアフロントにパーティションを追加して、サービスファイルを削除します
• Sparkが入力した元のデータの空のフィールドを処理します
• 単純な型を文字列にキャストする
• フィールド名を小文字に変換する
• Hiveでの個別のデータダンプとテーブル登録（DDL作成）
• Hiveと互換性がない可能性があるフィールド名をエスケープすることを忘れないでください
• Hiveでテーブルの登録を更新する方法を学ぶ

要約すると、ショーケースを構築するという決定は多くの落とし穴を隠していることに注意してください。したがって、実装で問題が発生した場合は、成功した専門知識を持つ経験豊富なパートナーに連絡することをお勧めします。



この記事を読んでいただきありがとうございます。この情報がお役に立てば幸いです。