Spark スキーマの進化の実践

読者の皆様、こんにちは！

この記事では、Neoflex のビッグ データ ソリューション ビジネス分野の主要コンサルタントが、Apache Spark を使用して可変構造のストアフロントを構築するためのオプションについて詳しく説明します。

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

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

この場合、対象となる店頭のレイアウトは事前に不明である。 さらに、スキームはデータに依存するため、事前に作成することはできません。また、この非常に弱い構造のデータを扱っているためです。

たとえば、今日、次の応答が記録されます。

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

そして明日、同じシステムから次の応答が返されます。

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

その結果、ストアフロントに別のフィールド「説明」が追加されるはずですが、それが追加されるかどうかは誰にもわかりません。

このようなデータにマートを作成するタスクはかなり標準的であり、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 を含むフォルダーと、このフォルダーに基づいて作成されたストアフロントがあります。 ソースからデータの次の部分をロードした後、ストアフロントには XNUMX 日分のデータが不足します。

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

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

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}

XNUMX つの異なるパーティションがあり、それぞれに XNUMX 行があるとします。
ソース データ全体を読み取ると、Spark は型を判断できるようになり、「a」が INT 型のネストされたフィールド「b」を持つ「構造体」型のフィールドであることを理解します。 ただし、各パーティションが個別に保存された場合は、互換性のないパーティション スキームを持つ寄木細工が作成されます。

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

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

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

この場合、寄木細工は一緒に読み取ることができるパーティションで構成されます。
実際にやったことのある人は苦笑いするでしょうが。 なぜ？ はい、おそらくあと 1 つの状況が発生する可能性が高いからです。 あるいは1.1つ。 あるいはXNUMXつ。 XNUMX つ目は、ほぼ確実ですが、JSON ファイルごとに数値型の見た目が異なるということです。 たとえば、{intField: XNUMX} や {intField: XNUMX} などです。 このようなフィールドが XNUMX つのバッチに表示される場合、スキーマのマージによってすべてが正しく読み取られ、最も正確な型が得られます。 ただし、異なる場合は、一方は intField: int になり、もう一方は intField: double になります。

この状況に対処するために、次のフラグがあります。

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

これで、パーティションが配置されたフォルダーができました。これを単一のデータフレームと店頭全体の有効な寄木細工に読み取ることができます。 はい？ いいえ。

テーブルを Hive に登録したことを覚えておく必要があります。 Hive ではフィールド名の大文字と小文字が区別されませんが、parquet では大文字と小文字が区別されます。 したがって、スキーマ field1: int と Field1: int を持つパーティションは、Hive では同じですが、Spark では同じではありません。 フィールド名を小文字に変更することを忘れないでください。

この後はすべて順調のようです。

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

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

現在、毎日、新しい寄木細工のパーティションがターゲットの店頭フォルダーに追加されており、その日の解析されたデータがそこに配置されているようです。 データ型が競合するパーティションがないことを事前に確認しました。

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

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

私たちは XNUMX 番目の問題に直面しています。 初めてテーブルを登録したときは、Spark に依存しました。 今はそれを自分たちで行うようになりましたが、寄木細工のフィールドは Hive で許可されていない文字で始まる可能性があることを覚えておく必要があります。 たとえば、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("配列<`", "配列<") つまり、次の代わりに安全な DDL を実行します。

create table tname (_field1 string, 1field string)

「_field1, 1field」のようなフィールド名を使用すると、フィールド名がエスケープされた安全な DDL が作成されます: create table `tname` (`_field1` string, `1field` string)。

疑問が生じます: 完全なスキーマを含むデータフレームを (PF コード内で) 正しく取得するにはどうすればよいでしょうか? このpfを入手するにはどうすればよいですか？ これが５つ目の問題です。 対象の店頭の寄木細工のファイルが含まれるフォルダーからすべてのパーティションの図を再度読み取りますか? この方法は最も安全ですが、困難です。

スキーマはすでに 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 でテーブル登録を更新する方法を学習します

要約すると、店舗を建設するという決定には多くの落とし穴が含まれていることがわかります。 したがって、実装で問題が発生した場合は、成功した専門知識を持つ経験豊富なパートナーに頼ることをお勧めします。

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

出所： habr.com