Spark DataFrameで効率的に列を連結：パフォーマンスを向上させるヒント

Apache Spark DataFrameで列を連結する

文字列型の列を結合する場合、concat関数とconcat_ws関数を用いることができます。

1 concat関数

concat関数は、複数の文字列型列をカンマ区切りで結合します。

from pyspark.sql.functions import concat

df = spark.createDataFrame([
    ("Alice", 20, "Seattle"),
    ("Bob", 30, "New York"),
    ("Charlie", 40, "Chicago"),
], ["name", "age", "city"])

# "name", "age", "city"列を結合して"full_info"列を作成
df_with_full_info = df.withColumn("full_info", concat(df["name"], lit(" "), df["age"], lit(", "), df["city"]))

df_with_full_info.show()

出力結果：

+-----+----+-----+------------+
|name |age | city| full_info   |
+-----+----+-----+------------+
|Alice|  20|Seattle| Alice 20, Seattle|
|Bob |  30|New York| Bob 30, New York|
|Charlie| 40|Chicago| Charlie 40, Chicago|
+-----+----+-----+------------+

concat_ws関数は、複数の文字列型列を結合し、区切り文字を指定することができます。

from pyspark.sql.functions import concat_ws

df = spark.createDataFrame([
    ("Alice", 20, "Seattle"),
    ("Bob", 30, "New York"),
    ("Charlie", 40, "Chicago"),
], ["name", "age", "city"])

# "name", "age", "city"列を結合して"full_info"列を作成し、区切り文字を"-"に指定
df_with_full_info = df.withColumn("full_info", concat_ws("-", df["name"], df["age"], df["city"]))

df_with_full_info.show()

+-----+----+-----+------------+
|name |age | city| full_info   |
+-----+----+-----+------------+
|Alice|  20|Seattle| Alice-20-Seattle|
|Bob |  30|New York| Bob-30-New York|
|Charlie| 40|Chicago| Charlie-40-Chicago|
+-----+----+-----+------------+

構造化型列を結合する場合、struct関数とwithColumn関数を用いることができます。

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

df1 = spark.createDataFrame([
    ("Alice", 20),
    ("Bob", 30),
    ("Charlie", 40),
], ["name", "age"])

df2 = spark.createDataFrame([
    ("Seattle"),
    ("New York"),
    ("Chicago"),
], ["city"])

# df1とdf2の列を構造化型列として結合して"profile"列を作成
profile_struct = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("city", StringType(), True),
])

df_with_profile = df1.withColumn("profile", struct(df1["name"], df1["age"], df2["city"]))

df_with_profile.show()

+-----+----+------------+
|name |age |      profile |
+-----+----+------------+
|Alice|  20|  Row(Alice, 20, Seattle)|
|Bob |  30|  Row(Bob, 30, New York)|
|Charlie| 40|  Row(Charlie, 40, Chicago)|
+-----+----+------------+

補足

• 上記の例では、Pythonを使用していますが、Scalaでも同様の操作が可能です。
• 列を結合する以外にも、文字列関数を用いて列を加工することもできます。詳細は、Apache Sparkドキュメント

Apache Spark DataFrame列の結合：サンプルコード

文字列型列の結合

from pyspark.sql.functions import concat

df = spark.createDataFrame([
    ("Alice", 20, "Seattle"),
    ("Bob", 30, "New York"),
    ("Charlie", 40, "Chicago"),
], ["name", "age", "city"])

# "name", "age", "city"列を結合して"full_info"列を作成
df_with_full_info = df.withColumn("full_info", concat(df["name"], lit(" "), df["age"], lit(", "), df["city"]))

df_with_full_info.show()

+-----+----+-----+------------+
|name |age | city| full_info   |
+-----+----+-----+------------+
|Alice|  20|Seattle| Alice 20, Seattle|
|Bob |  30|New York| Bob 30, New York|
|Charlie| 40|Chicago| Charlie 40, Chicago|
+-----+----+-----+------------+

from pyspark.sql.functions import concat_ws

df = spark.createDataFrame([
    ("Alice", 20, "Seattle"),
    ("Bob", 30, "New York"),
    ("Charlie", 40, "Chicago"),
], ["name", "age", "city"])

# "name", "age", "city"列を結合して"full_info"列を作成し、区切り文字を"-"に指定
df_with_full_info = df.withColumn("full_info", concat_ws("-", df["name"], df["age"], df["city"]))

df_with_full_info.show()

+-----+----+-----+------------+
|name |age | city| full_info   |
+-----+----+-----+------------+
|Alice|  20|Seattle| Alice-20-Seattle|
|Bob |  30|New York| Bob-30-New York|
|Charlie| 40|Chicago| Charlie-40-Chicago|
+-----+----+-----+------------+

構造化型列の結合

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

df1 = spark.createDataFrame([
    ("Alice", 20),
    ("Bob", 30),
    ("Charlie", 40),
], ["name", "age"])

df2 = spark.createDataFrame([
    ("Seattle"),
    ("New York"),
    ("Chicago"),
], ["city"])

# df1とdf2の列を構造化型列として結合して"profile"列を作成
profile_struct = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("city", StringType(), True),
])

df_with_profile = df1.withColumn("profile", struct(df1["name"], df1["age"], df2["city"]))

df_with_profile.show()

+-----+----+------------+
|name |age |      profile |
+-----+----+------------+
|Alice|  20|  Row(Alice, 20, Seattle)|
|Bob |  30|  Row(Bob, 30, New York)|
|Charlie| 40|  Row(Charlie, 40, Chicago)|
+-----+----+------------+

説明:

• サンプルコードでは、簡単な例を用いていますが、実際のデータ分析においては、結合する列や処理内容に応じて、より複雑なコードが必要となる場合があります。

上記以外にも、DataFrameの列を結合する方法はいくつかあります。

• crossJoin関数：すべての行を結合します。
• innerJoin関数：共通の列値を持つ行のみを結合します。
• leftJoin関数：左側のDataFrameのすべての行と、右

ここでは、Apache Spark DataFrame列を結合するその他の方法についていくつか紹介します。

SQL文を使用する

Spark DataFrameは、SQL文を使用して操作することもできます。結合操作の場合、以下のようなSQL文を使用できます。

• SELECT構文:

SELECT name, age, city,
       concat(name, " ", age, ", ", city) AS full_info
FROM df;

• FROM句の結合:

SELECT df1.name, df1.age, df2.city
FROM df1
JOIN df2
ON df1.name = df2.name;

RDDを使用する

Spark DataFrameは、RDD(Resilient Distributed Datasets)に変換することができます。RDDを使用して列を結合するには、以下のような操作を実行できます。

def combine_columns(df):
    # dfをRDDに変換
    rdd = df.rdd

    # 列を結合した新しいRDDを作成
    combined_rdd = rdd.map(lambda row: (row[0], row[1], row[2], row[0] + " " + str(row[1]) + ", " + row[2]))

    # 新しいRDDをDataFrameに変換
    combined_df = combined_rdd.toDF(["name", "age", "city", "full_info"])

    return combined_df

# 列を結合して新しいDataFrameを作成
combined_df = combine_columns(df)

ユーザー定義関数を使用する

独自の結合ロジックを実装したい場合は、ユーザー定義関数を作成することができます。

from pyspark.sql.functions import udf

def combine_columns_udf(name, age, city):
    return name + " " + str(age) + ", " + city

# ユーザー定義関数を登録
combine_columns_udf = udf(combine_columns_udf, StringType())

# ユーザー定義関数を使用して列を結合
df_with_full_info = df.withColumn("full_info", combine_columns_udf(df["name"], df["age"], df["city"]))

pyspark.sql.functionsモジュールには、列を操作するためのさまざまな関数が用意されています。結合操作以外にも、文字列の操作、数値演算、条件分岐など、さまざまな処理を実行することができます。

Apache Spark DataFrame列を結合するには、さまざまな方法があります。状況に応じて適切な方法を選択することが重要です。