注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。
PySpark の DataFrame は、SQL のテーブルが結合できるのと同様に、別のデータフレームまたは自身に結合することができます。データフレームは .join() メソッドを使用して他のデータフレームに結合されます。これには DataFrame 、結合する行の名前のような結合制約、およびメソッド（left 、 right 、 inner など）が必要です。
Copied!1 2 3# 左側のデータフレーム (df_left) と右側のデータフレーム (df_right) を、 # 'key' という共通のキーを使って左結合 (left join) し、その結果を df_joined に格納します。 df_joined = df_left.join(df_right, 'key', 'left')
df_joinedは、df_left.key == df_right.keyのleft結合の結果です。 PySparkは自動的にkey行のコピーの1つを削除するため、df_joinedにはkeyという名前の行が1つだけ含まれます。
df_leftとdf_rightで結合するキーが同じ名前を持っていない場合、結合を実行する前にそれらを最初にリネームすることを推奨します。
結合が完了すると衝突する可能性があるため、明示的に結合していないフィールドの名前を変更するか、または削除してください。 DataFrameのすべての行は、以下のようにループで特定のプレフィクスにリネームできます。
Copied!1 2 3 4# df.columns を通じて、データフレーム内の全ての列をループで処理します for column in df.columns: # withColumnRenamed メソッドを用いて、各列の名前を 'some_prefix_' + 元の列名 という新しい名前に変更します df = df.withColumnRenamed(column, 'some_prefix_' + column)
.join() メソッドは、単一のフィールドの代わりに結合するフィールドのリストを受け取ることができます。
Copied!1 2 3# 左結合（left join）を使って、df_leftとdf_rightを結合します。 # 結合キーは'column1'、'column2'、'column3'です。 df_joined = df_left.join(df_right, ['column1', 'column2', 'column3'], 'left')
df_joined は現在、column1、column2、column3 に対する結合となっています。これは、df_left と df_right の間で行名が一貫していることを前提としています。
PySpark は、論理演算子を使用して結合するための任意の式をサポートしています。例えば、行 ID で結合し、左の DataFrame の日付 start が右の DataFrame の日付 end よりも前であること、そして特定のフィールド X の内容によっては、右の DataFrame の Y が別の値を含んでいることを必要としないかもしれない、というような場合を考えてみましょう。
Copied!1 2 3 4 5 6 7 8 9 10 11# キー制約: df_leftのIDとdf_rightのIDが等しい key_constraint = df_left.ID == df_right.ID # 日付制約: df_leftの開始日がdf_rightの終了日より前 date_constraint = df_left.start < df_right.end # ケース制約: df_leftのXが'some_value'の場合、df_rightのYが'some_other_value'である必要があります case_constraint = F.when(df_left.X == 'some_value', df_right.Y == 'some_other_value')\ .otherwise(True) # すべての制約を組み合わせる combined_constraints = key_constraint & date_constraint & case_constraint # df_leftとdf_rightを制約に基づいて結合する df_joined = df_left.join(df_right, combined_constraints, 'left')
キーまたはその他の制約によるマッチングなしで、2つのデータフレーム間の行のすべての組み合わせを生成するために、クロス結合（デカルト積とも呼ばれる）を使用します。クロス結合は、メモリやパフォーマンスの問題を引き起こすリスクがあるため、可能な限り避けるべきです。
結果をすぐにフィルター処理するつもりの場合は、クロス結合を使用しないでください。代わりに、フィルター条件を結合制約に組み込んで、より効率的な解決策を実現してください（上記の高度な任意の結合制約を参照）。
クロス結合を使用するには、ユーザーのコードリポジトリで CROSS_JOIN_ENABLED プロファイルを明示的にインポートする必要があります。
Copied!1 2 3 4 5 6 7 8 9 10from transforms.api import configure # プロファイルに "CROSS_JOIN_ENABLED" を設定します @configure(profile=["CROSS_JOIN_ENABLED"]) @transform_df( ... ) # my_compute_function という関数を定義します。この関数は、二つの入力データフレーム (input_a, input_b) を受け取り、それらをクロス結合します。 def my_compute_function(input_a, input_b): return input_a.crossJoin(input_b)