注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。
このセクションでは、インクリメンタル（差分処理）で計算できるトランスフォームの幅広い例を紹介します。
これらの例では、students と students_updated の2つの入力を使用して、インクリメンタル計算を説明します。students 入力には3人の学生が含まれており、インクリメンタルではありません。つまり、履歴がありません。
>>> students.dataframe('previous').sort('id').show()
+---+----+---+---+
| id|hair|eye|sex|
+---+----+---+---+
+---+----+---+---+
>>>
>>> students.dataframe('current').sort('id').show()
+---+-----+-----+------+
| id| hair| eye| sex|
+---+-----+-----+------+
| 1|Brown|Green|Female|
| 2| Red| Blue| Male|
| 3|Blond|Hazel|Female|
+---+-----+-----+------+
>>>
>>> students.dataframe('added').sort('id').show()
+---+-----+-----+------+
| id| hair| eye| sex|
+---+-----+-----+------+
| 1|Brown|Green|Female|
| 2| Red| Blue| Male|
| 3|Blond|Hazel|Female|
+---+-----+-----+------+
>>>
>>> # 入力のデフォルトの読み取りモードは 'added' であることを思い出してください。
>>> students.dataframe('added') is students.dataframe()
True
students_updated 入力は students と同じですが、さらに3人の学生を含む更新が追加されています。この更新によって入力がインクリメンタルになります。そのため、previous DataFrame は空ではありません。
>>> students_updated.dataframe('previous').sort('id').show()
+---+-----+-----+------+
| id| hair| eye| sex|
+---+-----+-----+------+
| 1|Brown|Green|Female|
| 2| Red| Blue| Male|
| 3|Blond|Hazel|Female|
+---+-----+-----+------+
>>>
>>> students_updated.dataframe('current').sort('id').show()
+---+-----+-----+------+
| id| hair| eye| sex|
+---+-----+-----+------+
| 1|Brown|Green|Female|
| 2| Red| Blue| Male|
| 3|Blond|Hazel|Female|
| 4|Brown|Green|Female|
| 5|Brown| Blue| Male|
| 6|Blond|Hazel|Female|
+---+-----+-----+------+
>>>
>>> students_updated.dataframe('added').sort('id').show()
+---+-----+-----+------+
| id| hair| eye| sex|
+---+-----+-----+------+
| 4|Brown|Green|Female|
| 5|Brown| Blue| Male|
| 6|Blond|Hazel|Female|
+---+-----+-----+------+
>>>
>>> # 入力のデフォルトの読み取りモードは 'added' であることを思い出してください。
>>> students_updated.dataframe('added') is students_updated.dataframe()
True
追加のみのインクリメンタル計算では、追加される出力行が、追加された入力行だけによって決まります。つまり、出力を計算するために、トランスフォームは次の処理を行います。
列のデータ型の変更、日付の文字列への書式設定、絞り込みは、いずれも追加のみの計算の例です。これらの例では、追加された入力行をそれぞれ加工または削除して、出力行を生成します。
追加のみのトランスフォームをインクリメンタルにするために必要な変更は、incremental() デコレーターだけであることに注目してください。
インクリメンタルに実行する場合、デフォルトの読み取りモード added では、トランスフォームは新しい学生だけを読み取ります。また、デフォルトの書き込みモード modify では、トランスフォームは絞り込んだ新しい学生だけを出力に追加します。
非インクリメンタルに実行する場合、デフォルトの読み取りモード added では、トランスフォームは入力全体を読み取ります。また、デフォルトの書き込みモード replace では、トランスフォームは出力を、絞り込んだ学生全体で置き換えます。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13import polars as pl from transforms.api import transform, incremental, Input, Output @incremental() @transform.using( students=Input('/examples/students_hair_eye_color'), processed=Output('/examples/hair_eye_color_processed') ) def incremental_filter(students, processed): new_students_df = students.polars() processed.write_table( new_students_df.filter(pl.col('hair') == 'Brown') )
Copied!1 2 3 4 5 6 7 8 9 10 11 12from transforms.api import transform, incremental, Input, Output @incremental() @transform.using( students=Input('/examples/students_hair_eye_color'), processed=Output('/examples/hair_eye_color_processed') ) def incremental_filter(students, processed): new_students_df = students.pandas() processed.write_table( new_students_df[new_students_df['hair'] == 'Brown'] )
Copied!1 2 3 4 5 6 7 8 9 10 11 12from transforms.api import transform, incremental, Input, Output @incremental() @transform( students=Input('/examples/students_hair_eye_color'), processed=Output('/examples/hair_eye_color_processed') ) def incremental_filter(students, processed): new_students_df = students.dataframe() processed.write_dataframe( new_students_df.filter(new_students_df.hair == 'Brown') )
更新をインクリメンタルに計算するために、トランスフォームで以前の出力を参照する必要がある場合があります。その例が distinct() ↗ メソッドです。
トランスフォームで重複行を削除するには（現在の出力が正しいと仮定すると）、入力の新しい行から重複を除去し、その行が出力にすでに存在していないことを確認する必要があります。
Copied!1 2 3 4 5 6 7 8 9 10 11 12@incremental() @transform( students=Input('/examples/students_hair_eye_color'), processed=Output('/examples/hair_eye_color_processed') ) def incremental_distinct(students, processed): new_students_df = students.dataframe() processed.write_dataframe( new_students_df.distinct().subtract( processed.dataframe('previous', schema=new_students_df.schema) ) )
ここでは、出力データセットで previous 読み取りモードを使用します。これにより、前回のビルドで出力された DataFrame ↗ が返されます。previous 出力が存在しない可能性があるため、空の DataFrame を正しく構築できるように、dataframe('previous') 呼び出しにスキーマを指定する必要があります。
常に出力全体を置き換えるトランスフォームもあります。それでも、このようなトランスフォームでもインクリメンタル計算の利点を活用できることが多くあります。その例の1つが統計の集計です。たとえば、列のそれぞれの一意な値が出現する回数を数える処理です。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23from pyspark.sql import functions as F @incremental() @transform( students=Input('/examples/students_hair_eye_color'), processed=Output('/examples/hair_eye_color_processed') ) def incremental_group_by(students, processed): # 新しい学生のみを対象に、髪の色ごとの件数を計算します。 new_hair_counts_df = students.dataframe().groupBy('hair').count() # 以前の件数との和集合を作成します。 out_schema = new_hair_counts_df.schema all_counts_df = new_hair_counts_df.union( processed.dataframe('previous', schema=out_schema) ) # 髪の色でグループ化し、2つの件数の集合を合計します。 totals_df = all_counts_df.groupBy('hair').agg(F.sum('count').alias('count')) # 出力を完全に置き換えるため、出力モードを常に 'replace' に設定します。 processed.set_mode('replace') processed.write_dataframe(totals_df.select(out_schema.fieldNames()))
ここでも、previous 出力が存在しない可能性があるため、空の DataFrame を正しく構築できるように、dataframe('previous') 呼び出しにスキーマを指定する必要があります。
インクリメンタルトランスフォームでは、スキーマのない入力データセットに追加されたファイルから Spark DataFrame を作成し、その DataFrame の内容を出力に追加する必要がある場合があります。このようなトランスフォームには、2つの実装パターンがあります。
期待されるスキーマを静的に指定し、生成される DataFrame がそのスキーマに従うようにします（入力データ内の余分なフィールドを無視する、入力データにないフィールドに null を補う、など）。
入力データに存在するフィールドを動的に取り込みます。
動的に取り込む実装パターンでは、出力に追加する DataFrame のスキーマが既存の出力に_対応_していることを確認する必要があります。スキーマが既存の出力に対応するには、次の条件を満たす必要があります。
新しいデータ内の列で、既存の出力内の列と同じ名前を持つものは、データ型も同じである必要があります。
既存の出力内のすべての列が、新しいデータにも存在する必要があります。
新しいデータには、既存のデータ内の列と名前の大文字・小文字だけが異なる列があってはなりません。たとえば、既存のデータにすでに value 列がある場合、新しいデータに Value 列があってはなりません。
これらの条件が満たされていることを確認するには、既存のデータのスキーマを動的に調べる必要があります。つまり、スキーマを指定せずに dataframe('previous') を呼び出します。これは、トランスフォームがインクリメンタルに実行されている場合に限りサポートされます。
以下のコードは、これらの原則をすべて組み合わせたものです。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59from functools import reduce from pyspark.sql import functions as F from transforms.api import transform, incremental, Input, Output from transforms.verbs.dataframes import sanitize_schema_for_parquet @incremental() @transform( csvs=Input('/examples/dataset_of_csvs'), parsed=Output('/examples/parsed_csvs') ) def incremental_read_csv(ctx, csvs, parsed): input_fs = csvs.filesystem() def process_file(f): df = ( ctx.spark_session.read # すべての列を文字列にするため、inferSchema を False に設定します。 # これにより、ファイル間で推論結果が一貫しないことによる問題を防ぎます。 # 同一のインクリメンタルビルド内でも、異なるビルド間でも有効です。 .option("inferSchema", False) .option("header", True) .csv(input_fs.hadoop_path + "/" + f) ) sanitized = sanitize_schema_for_parquet(df) # ファイル間で大文字と小文字の使い方が一貫しないことによる問題を防ぐため、すべての列名を小文字にします。 # 同一のインクリメンタルビルド内でも、異なるビルド間でも有効です。 # このロジックでは、1つのファイルに大文字と小文字だけが異なる列が含まれる場合には # 対応できないことに注意してください。 return sanitized.select(*(F.col(a).alias(a.lower()) for a in sanitized.columns)) dfs = [process_file(f.path) for f in input_fs.ls()] if len(dfs) == 0: parsed.abort() return df = reduce( lambda a, b: a.unionByName(b, allowMissingColumns=True), dfs[1:], dfs[0], ) if not ctx.is_incremental: parsed.write_dataframe(df) else: existing_columns = parsed.dataframe("previous").columns columns_in_new_data = set(df.columns) resolved_schema_df = df.select( *df.columns, *( F.lit(None).cast("string").alias(col) for col in existing_columns if col not in columns_in_new_data ) ) parsed.write_dataframe(resolved_schema_df)
顧客が送信した注文を記録する Orders と、完了した配送を記録する Deliveries の2個のテーブルがあり、商品の配送にかかる時間を含む DeliveryDuration テーブルを計算したいとします。Orders と Deliveries の両テーブルでは、新しい行が追加されるだけで、行が変更されることはありません。それでも、この2つのインクリメンタルデータセットを単純に結合しても機能しません。たとえば、Orders テーブルには、Deliveries テーブルにまだ存在しない orderIds が含まれている場合があります。
Orders: Deliveries:
+---------+---------------+ +---------+--------------+ +---------+------------------+
| orderId | submittedDate | | orderId | deliveryDate | | orderId | deliveryDuration |
+---------+---------------+ +---------+--------------+ ----> +---------+------------------+
| 1001 | 2019-08-21 | join on | 1001 | 2019-08-23 | | 1001 | 2 |
+---------+---------------+ orderId +---------+--------------+ +---------+------------------+
| 1002 | 2019-08-22 |
+---------+---------------+
| 1003 | 2019-08-23 |
+---------+---------------+
Orders と Deliveries の両テーブルで orderId が厳密に増加すると仮定すると、deliveryDuration を計算した最後の orderId（maxComputedOrderId）を確認し、Orders と Deliveries の両テーブルから、orderId が maxComputedOrderId より大きい行だけを取得できます。
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41from transforms.api import transform, Input, Output, incremental from pyspark.sql import types as T from pyspark.sql import functions as F @incremental(snapshot_inputs=['orders', 'deliveries']) @transform( orders=Input('/example/Orders'), deliveries=Input('/example/Deliveries'), delivery_duration=Output('/example/New_Delivery_Date') ) def compute_delivery_duration(orders, deliveries, delivery_duration): def to_fields(datatype, names, nullable=True): return [T.StructField(n, datatype, nullable) for n in names] # deliveryDuration に渡すスキーマを生成します。 fields = to_fields(T.IntegerType(), ['orderId', 'deliveryDuration']) # 以前のバージョンのスキーマを参照できないため、スキーマを明示的に定義します。 maxComputedOrderId = ( delivery_duration .dataframe('previous', schema=T.StructType(fields)) .groupby() .max('orderId') .collect()[0][0] ) # 最初の反復では、delivery_duration データセットがまだ存在しないため、maxComputedOrderId は空です。 if maxComputedOrderId == None: maxComputedOrderId = 0 ordersNotProcessed = orders.dataframe().filter(F.col('orderId') > maxComputedOrderId) deliveriesNotProcessed = deliveries.dataframe().filter(F.col('orderId') > maxComputedOrderId) newDurations = ( ordersNotProcessed .join(deliveriesNotProcessed, 'orderId', how='left') .withColumn('deliveryDuration', F.datediff(F.col('deliveryDate'), F.col('submittedDate'))) .drop(*['submittedDate', 'deliveryDate']) ) delivery_duration.write_dataframe(newDurations)
今後、インクリメンタルデータセットに別の列を追加したいとします。出力に別の列を追加しても is_incremental フラグは無効にならないため、次の実行では新しい行が計算され、新しい列を含むデータが書き込まれます。以前に書き込まれたすべての行では、この列は null になります。
ただし、以前の行にもこの列の値を設定したい場合があります。トランスフォームの semantic_version を上げると、1回だけ非インクリメンタルに実行されます。読み取りモード「added」を使用している場合、入力にはすべてのデータが含まれるため、再計算して新しい列を追加できます。
トランスフォームでスナップショット入力から履歴データセットを作成していた場合は、以前のデータが入力の SNAPSHOT トランザクションの積み重ねに含まれているため、少し複雑になります。この場合は、Palantir の担当者にお問い合わせください。
この例では新しい列の追加について説明しましたが、上記の考え方はあらゆる種類のロジック変更に当てはまります。
新しいブランチを作成してそこでビルドを実行すると、ビルドはインクリメンタルに実行されます。新しいブランチでの最初のビルドでは、このブランチの作成元となったブランチで最後にコミットされたトランザクションが、前回のトランザクションとして扱われます。
次の方法でデータの差分だけを処理する方法を説明しました。
また、次の方法も探索しました。
入力データセットに多数のトランザクションが蓄積されている場合、または SNAPSHOT モードで実行される場合は、トランザクション制限を設定して、各ジョブが読み取るデータを制限します。詳細については、インクリメンタル入力のバッチサイズを制限するを参照してください。
インクリメンタルのエラーを理解するには、トランザクションとデータセットビューの概念を読んでおくと理解しやすくなります。また、場合によってはこれらを読んでおく必要があります。
ジョブがインクリメンタルに実行される場合、そのインクリメンタル入力データセットには、データセットビュー全体ではなく、未処理のトランザクション範囲のみが含まれます。
データセットに次のトランザクション履歴があるとします。
SNAPSHOT (1) --> UPDATE (2) --> UPDATE (3) --> UPDATE (4) --> UPDATE (5)
|
Last processed transaction
データセットが前回ビルドされた時点で、最新のトランザクションは (3) でした。その後、トランザクション (4) と (5) がコミットされたため、未処理のトランザクション範囲は (4) — (5) です。
データセットのビューは、トランザクション範囲 (1) — (5) です。ビューの「最上位」にあるトランザクション（最新のトランザクション）は、ブランチの HEAD と呼ばれることがあります（これも git との類推によるものです）。git と同様に、ブランチはトランザクションへのポインターであるため、ブランチはトランザクション (5) を指すと表現します。複数のブランチがそれぞれ異なるトランザクションを指すことができ、ブランチ間でトランザクション履歴を共有している場合もあります。
SNAPSHOT (1) ─> UPDATE (2) ─> UPDATE (3) ─> UPDATE (4) ─> UPDATE (5) [develop]
|
└─> UPDATE [feature-branch]
Catalog:TransactionsNotInViewジョブをインクリメンタルに実行するために、ジョブの開始時に一連のチェックが実行されます。 そのうちの1つは、未処理のトランザクション範囲が厳密にインクリメンタルであること（つまり、ファイルの変更が追加のみであること。インクリメンタル計算の要件を参照）を確認します。このチェックでは、未処理のトランザクション範囲と処理済みのトランザクション範囲のファイルを比較します。
ただし、ブランチの HEAD が移動されている場合、インクリメンタルジョブは不整合な状態になります。両方の範囲を比較する意味がなくなるため、Catalog:TransactionNotInView エラーがスローされます。
このエラーが発生する状況を示す図を以下に示します。
SNAPSHOT (1) ─> UPDATE (2) ─> UPDATE (3) ─> UPDATE (4) ─> UPDATE (5)
| (last processed (branch's previous
| transaction) HEAD, now orphan)
|
└─> UPDATE (6) --> UPDATE (7, branch's current HEAD)
ここでは、処理済みのトランザクション範囲は (1) — (5)、現在のブランチの HEAD は (7) を指し、現在のビューはトランザクション (1)、(2)、(6)、(7) で構成されています。
処理済みのトランザクションの一部がブランチの HEAD の上流にないため、これは不整合な状態です。実際、(3) は上流にありません。つまり、以前の HEAD (3) が現在のビューに含まれなくなったため、Catalog:TransactionNotInView がスローされます。
Catalog:InconsistentBranchesOrderスローされる可能性があるもう1つの Catalog エラーは Catalog:InconsistentBranchesOrder で、最後に処理されたトランザクション（prevTransaction）がブランチの HEAD（endTransaction）より大きい場合に発生します。これは、データセットの HEAD が前回のトランザクションより前のトランザクションに移動された場合に発生する可能性があります。
このエラーが発生する状況を示す図を以下に示します。
SNAPSHOT (1) --> UPDATE (2) --> UPDATE (3) --> UPDATE (4) --> UPDATE (5)
| |
Current HEAD Last processed transaction
ブランチの HEAD が変更される理由は2つあります。
SNAPSHOT であるため、不整合な状態にはなりません。これに対処するには、次のいずれかを行う必要があります。
parentRef として、updateBranch2 エンドポイントを使用します。このエンドポイントは、経験豊富なユーザーのみが使用してください。