構文チートシート

PySpark SQL でよく使われるパターンと関数のクイックリファレンスです。

• 一般的なパターン
  • ログ出力
  • 関数とデータ型をインポートする
  • 絞り込み
  • 結合
  • 列の操作
  • キャスト、null 値の補完、重複の処理
• 文字列の操作
  • 文字列フィルター
  • 文字列関数
• 数値の操作
• 日付とタイムスタンプの操作
• 配列の操作
• 集計操作
• ウィンドウ操作
• 高度な操作
  • 再パーティション化する
  • UDF（ユーザー定義関数）

探している情報が見つからない場合は、PySpark 公式ドキュメント ↗に記載されている可能性があります。

一般的なパターン

ログ出力

Copied!
1
2
3
4
5
6
7
# Code Workbook 内
print("example log output")

# Code Repositories 内
import logging
logger = logging.getLogger(__name__)
logger.info("example log output")

関数とデータ型をインポートする

Copied!
1
2
# 以下では F.my_function() および T.my_type() として簡単に参照できます
from pyspark.sql import functions as F, types as T

絞り込み

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 等しい条件で絞り込みます
df = df.filter(df.is_adult == 'Y')

# >, <, >=, <= の条件で絞り込みます
df = df.filter(df.age > 25)

# 複数の条件を指定する場合は、それぞれの条件を括弧で囲む必要があります
df = df.filter((df.age > 25) & (df.is_adult == 'Y'))

# 許容される値のリストと比較します
df = df.filter(col('first_name').isin([3, 4, 7]))

# 結果を並べ替えます
df = df.orderBy(df.age.asc())
df = df.orderBy(df.age.desc())

結合

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
# 別のデータセットを左結合します
df = df.join(person_lookup_table, 'person_id', 'left')

# 別のデータセットを左アンチ結合します（左側のデータフレームで一致しない行を返します）
df = df.join(person_lookup_table, 'person_id', 'leftanti');

# 左側と右側のデータセットの異なる列で照合します
df = df.join(other_table, df.id == other_table.person_id, 'left')

# 複数の列で照合します
df = df.join(other_table, ['first_name', 'last_name'], 'left')

# クロス結合（直積）
# CROSS_JOIN_ENABLED プロファイルが必要です
from transforms.api import configure

@configure(profile=['CROSS_JOIN_ENABLED'])
def my_compute_function(input_a, input_b):
    return input_a.crossJoin(input_b)

# 1行で参照コードを結合する場合に便利です
def lookup_and_replace(df1, df2, df1_key, df2_key, df2_value):
    return (
        df1
        .join(df2[[df2_key, df2_value]], df1[df1_key] == df2[df2_key], 'left')
        .withColumn(df1_key, F.coalesce(F.col(df2_value), F.col(df1_key)))
        .drop(df2_key)
        .drop(df2_value)
    )

df = lookup_and_replace(people, pay_codes, id, pay_code_id, pay_code_desc)

列の操作

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
# 新しい固定値の列を追加します
df = df.withColumn('status', F.lit('PASS'))

# 新しい動的な列を作成します
df = df.withColumn('full_name', F.when(
    (df.fname.isNotNull() & df.lname.isNotNull()), F.concat(df.fname, df.lname)
).otherwise(F.lit('N/A')))

# 保持する列を選択し、必要に応じて一部の列の名前を変更します
df = df.select(
    'name',
    'age',
    F.col('dob').alias('date_of_birth'),
)

# 列を削除します
df = df.drop('mod_dt', 'mod_username')

# 列の名前を変更します
df = df.withColumnRenamed('dob', 'date_of_birth')

# 別のデータセットにも存在する列をすべて保持します
df = df.select(*(F.col(c) for c in df2.columns))

# 列の名前を一括で変更・クリーンアップする
for col in df.columns:
    df = df.withColumnRenamed(col, col.lower().replace(' ', '_').replace('-', '_'))

キャスト、null 値の補完、重複の処理

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 列を別のデータ型にキャストします
df = df.withColumn('price', df.price.cast(T.DoubleType()))

# すべての null を特定の値に置き換えます
df = df.fillna({
    'first_name': 'Tom',
    'age': 0,
})

# null ではない最初の値を取得します
df = df.withColumn('last_name', F.coalesce(df.last_name, df.surname, F.lit('N/A')))

# データセット内の重複する行を削除します（distinct() と同じです）
df = df.dropDuplicates()

# 特定の列のみを考慮して、重複する行を削除します
df = df.dropDuplicates(['name', 'height'])

文字列の操作

文字列フィルター

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 含む - col.contains(string)
df = df.filter(df.name.contains('o'))

# 指定した文字列で始まる - col.startswith(string)
df = df.filter(df.name.startswith('Al'))

# 指定した文字列で終わる - col.endswith(string)
df = df.filter(df.name.endswith('ice'))

# null である - col.isNull()
df = df.filter(df.is_adult.isNull())

# null ではない - col.isNotNull()
df = df.filter(df.first_name.isNotNull())

# パターンに一致する - col.like(string_with_sql_wildcards)
df = df.filter(df.name.like('Al%'))

# 正規表現に一致する - col.rlike(regex)
df = df.filter(df.name.rlike('[A-Z]*ice$'))

# リストに含まれる - col.isin(*values)
df = df.filter(df.name.isin('Bob', 'Mike'))

文字列関数

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
# 部分文字列 - col.substr(startPos, length)（1から始まるインデックス）
df = df.withColumn('short_id', df.id.substr(1, 10))

# トリム - F.trim(col)
df = df.withColumn('name', F.trim(df.name))

# 左側のパディング - F.lpad(col, len, pad)
# 右側のパディング - F.rpad(col, len, pad)
df = df.withColumn('id', F.lpad('id', 4, '0'))

# 左側のトリム - F.ltrim(col)
# 右側のトリム - F.rtrim(col)
df = df.withColumn('id', F.ltrim('id'))

# 連結 - F.concat(*cols)（いずれかの列が null の場合は null）
df = df.withColumn('full_name', F.concat('fname', F.lit(' '), 'lname'))

# 区切り文字付きの連結 - F.concat_ws(delimiter, *cols)（null を無視します）
df = df.withColumn('full_name', F.concat_ws('-', 'fname', 'lname'))

# 正規表現による置換 - F.regexp_replace(str, pattern, replacement)
df = df.withColumn('id', F.regexp_replace(id, '0F1(.*)', '1F1-$1'))

# 正規表現による抽出 - F.regexp_extract(str, pattern, idx)
df = df.withColumn('id', F.regexp_extract(id, '[0-9]*', 0))

数値の操作

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 四捨五入 - F.round(col, scale=0)
df = df.withColumn('price', F.round('price', 0))

# 切り下げ - F.floor(col)
df = df.withColumn('price', F.floor('price'))

# 切り上げ - F.ceil(col)
df = df.withColumn('price', F.ceil('price'))

# 絶対値 - F.abs(col)
df = df.withColumn('price', F.abs('price'))

# X の Y 乗 – F.pow(x, y)
df = df.withColumn('exponential_growth', F.pow('x', 'y'))

# 複数の列から最小値を選択します – F.least(*cols)
df = df.withColumn('least', F.least('subtotal', 'total'))

# 複数の列から最大値を選択します – F.greatest(*cols)
df = df.withColumn('greatest', F.greatest('subtotal', 'total'))

日付とタイムスタンプの操作

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
# 形式が既知の文字列を日付に変換します（時刻情報を含みません）
df = df.withColumn('date_of_birth', F.to_date('date_of_birth', 'yyyy-MM-dd'))

# 形式が既知の文字列をタイムスタンプに変換します（時刻情報を含みます）
df = df.withColumn('time_of_birth', F.to_timestamp('time_of_birth', 'yyyy-MM-dd HH:mm:ss'))

# 日付から年を取得します:       F.year(col)
# 日付から月を取得します:      F.month(col)
# 日付から日を取得します:        F.dayofmonth(col)
# 日付から時を取得します:       F.hour(col)
# 日付から分を取得します:     F.minute(col)
# 日付から秒を取得します:     F.second(col)
df = df.filter(F.year('date_of_birth') == F.lit('2017'))

# 日数を加算・減算します
df = df.withColumn('three_days_after', F.date_add('date_of_birth', 3))
df = df.withColumn('three_days_before', F.date_sub('date_of_birth', 3))

# 月数を加算・減算します
df = df.withColumn('next_month', F.add_months('date_of_birth', 1))
df = df.withColumn('previous_month', F.add_months('date_of_birth', -1))

# 2つの日付の間の日数を取得します
df = df.withColumn('days_between', F.datediff('end', 'start'))

# 2つの日付の間の月数を取得します
df = df.withColumn('months_between', F.months_between('end', 'start'))

# date_of_birth が 2017-05-10 から 2018-07-21 の間にある行のみを保持します
df = df.filter(
    (F.col('date_of_birth') >= F.lit('2017-05-10')) &
    (F.col('date_of_birth') <= F.lit('2018-07-21'))
)

配列と構造体の操作

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 列の配列 - F.array(*cols)
df = df.withColumn('full_name', F.array('fname', 'lname'))

# 空の配列 - F.array(*cols)
df = df.withColumn('empty_array_column', F.array(F.lit("")))

# 既存の列から配列または構造体の列を作成します
df = df.withColumn('guardians', F.array('guardian_1', 'guardian_2'))
df = df.withColumn('properties', F.struct('hair_color', 'eye_color'))

# 配列の列からインデックスで、またはマップの列からキーで抽出します（無効な場合は null）
df = df.withColumn('first_guardian', F.element_at(F.col('guardians'), 1))

# 構造体の列からフィールド名で抽出します
df = df.withColumn('hair_color', F.col('properties.hair_color'))

# 配列または構造体の列を複数の行に展開します
df = df.select(F.col('child_name'), F.explode(F.col('guardians')))
df = df.select(F.col('child_name'), F.explode(F.col('properties')))

# 構造体の列を複数の列に展開します
df = df.select(F.col('child_name'), F.col('properties.*'))

集計操作

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 行の件数:                F.count(*cols), F.countDistinct(*cols)
# グループ内の行の合計:     F.sum(*cols)
# グループ内の行の平均:    F.mean(*cols)
# グループ内の行の最大値:     F.max(*cols)
# グループ内の行の最小値:     F.min(*cols)
# グループ内の最初の行:       F.first(*cols, ignorenulls=False)
df = df.groupBy(col('address')).agg(
  count('uuid').alias('num_residents'),
  max('age').alias('oldest_age'),
  first('city', True).alias('city')
)

# グループ内のすべての行を集合にまとめます:       F.collect_set(col)
# グループ内のすべての行をリストにまとめます:      F.collect_list(col)
df = df.groupBy('address').agg(F.collect_set('name').alias('resident_names'))

ウィンドウ操作

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from pyspark.sql.window import Window

# パーティション単位の計算と順序に基づく計算のウィンドウを定義します
customer_window = Window.partitionBy('customer')
ordered_customer_window = customer_window.orderBy('spend')

# 顧客ごとのパーティション全体に集計関数を適用します
df = df.withColumn('average_spend', F.avg('spend').over(customer_window))
df = df.withColumn('max_spend', F.max('spend').over(customer_window))

# 並べ替え済みのパーティション内で順位付け関数を適用します
df = df.withColumn('rank', F.rank().over(ordered_customer_window))
df = df.withColumn('dense_rank', F.dense_rank().over(ordered_customer_window))
df = df.withColumn('row_number', F.row_number().over(ordered_customer_window))

# 並べ替え済みのパーティション内の次の値と前の値にアクセスします
df = df.withColumn('next_value', F.lead('spend', 1).over(ordered_customer_window))
df = df.withColumn('prev_value', F.lag('spend', 1).over(ordered_customer_window))

高度な操作

再パーティション化する

Copied!
1
2
# パーティションを再分割 – df.repartition(num_output_partitions)
df = df.repartition(1)

UDF（ユーザー定義関数）

Copied!
1
2
3
4
5
6
7
8
9
# 各行の age 列に2を掛けます
times_two_udf = F.udf(lambda x: x * 2)
df = df.withColumn('age', times_two_udf(df.age))

# 行の name として使用する値をランダムに選択します
import random

random_name_udf = F.udf(lambda: random.choice(['Bob', 'Tom', 'Amy', 'Jenna']))
df = df.withColumn('name', random_name_udf())