注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。
Java Spark SQL でよく使われるパターンと関数のクイックリファレンスガイドです。
Java Spark の詳細については、Java Spark 公式ドキュメント ↗を参照してください。
Copied!1 2 3 4 5 6 7import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class MyClass { private static final Logger LOG = LoggerFactory.getLogger(MyClass.class); LOG.info("example log output"); }
Copied!1 2 3 4import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import static org.apache.spark.sql.functions.*; import org.apache.spark.sql.types.DataTypes;
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18// 等しい条件で絞り込みます df = df.filter(col("is_adult").equalTo("Y")); // >, <, >=, <= の条件で絞り込みます df = df.filter(col("age").gt(25)); df = df.filter(col("age").lt(25)); df = df.filter(col("age").geq(25)); df = df.filter(col("age").leq(25)); // 複数の条件 df = df.filter(col("age").gt(25).and(col("is_adult").equalTo("Y"))); // 許可された値のリストと比較します df = df.filter(col("first_name").isin(List.of(3, 4, 7))); // 結果を並べ替えます df = df.orderBy(col("age").asc()); df = df.orderBy(col("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// 別のデータセットを左結合します df = df.join(personLookupTable, col("person_id"), "left"); // 別のデータセットを左アンチ結合します（左側のデータフレームで一致しない行を返します） df = df.join(personLookupTable, col("person_id"), "leftanti"); // 左側と右側のデータセットの異なる列で照合します df = df.join(otherTable, col("id").equalTo(col("person_id")), "left"); // 複数の列で照合します df = df.join(otherTable, col("first_name").equalTo(col("name")).and( col("last_name").equalTo(col("family_name"))), "left"); // コードの参照用の結合を1行で記述するのに便利です public Dataset<Row> lookupAndReplace(Dataset<Row> df1, Dataset<Row> df2, String df1Key, String df2Key, String df2Value) { return df1.join(df2.select(col(df2Key), col(df2Value)), col(df1Key).equalTo(col(df2Key)), "left") .withColumn(df1Key, coalesce(col(df2Value), col(df1Key))) .drop(df2Key, df2Value); } Dataset<Row> df = lookupAndReplace(people, payCodes, id, payCodeId, payCodeDesc);
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// 固定値を持つ新しい列を追加します df = df.withColumn("status", lit("PASS")); // 新しい動的な列を作成します df = df.withColumn("full_name", when( (col("fname").isNotNull().or(col("lname").isNotNull())), concat_ws(" ", col("fname"), col("lname"))) .otherwise(lit(null))); // 保持する列を選択し、必要に応じて一部の列の名前を変更します df = df.select( col("name"), col("age"), col("dob").alias("date_of_birth") ); // 列を削除します df = df.drop("mod_dt", "mod_username"); // 列の名前を変更します df = df.withColumnRenamed("dob", "date_of_birth"); // 別のデータセットにも存在する列をすべて保持します import org.apache.spark.sql.Column; import scala.collection.JavaConversions; List<Column> columnsToSelect = new ArrayList<Column>(); List<String> df2Columns = Arrays.asList(df2.columns()); for (String c : df.columns()) { if (df2Columns.contains(c)) { columnsToSelect.add(col(c)); } } df = df.select(JavaConversions.asScalaBuffer(columnsToSelect)); // 列の名前を一括で変更・整形します for (String c : df.columns()) { df = df.withColumnRenamed(c, c.toLowerCase().replace(" ", "_").replace("-", "_")); }
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21import org.apache.spark.sql.types.DataTypes; // 列を別のデータ型にキャストします（どちらの方法も同等です） df = df.withColumn("price", col("price").cast(DataTypes.DoubleType)); df = df.withColumn("price", col("price").cast("double")); // すべての値が null の列を作成し、特定のデータ型にキャストします df = df.withColumn("name", lit(null).cast(DataTypes.StringType)); df = df.withColumn("name", lit(null).cast("string")); // すべての null を特定の値に置き換えます df = df.na().fill(ImmutableMap.of("first_name", "Tom", "age", 0)); // null ではない最初の値を取得します df = df.withColumn("last_name", coalesce(col("last_name"), col("surname"), 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(col("name").contains("o")); //で始まる - col.startsWith(string) df = df.filter(col("name").startsWith("Al")); // で終わる - col.endsWith(string) df = df.filter(col("name").endsWith("ice")); // null である - col.isNull() df = df.filter(col("is_adult").isNull()); // null ではない - col.isNotNull() df = df.filter(col("first_name").isNotNull()); // パターンに一致する - col.like(string_with_sql_wildcards) df = df.filter(col("name").like("Al%")); // 正規表現に一致する - col.rlike(regex) df = df.filter(col("name").rlike("[A-Z]*ice$")); // リストに含まれる - col.isin(Object... values) df = df.filter(col("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", col("id").substr(1, 10)); // 前後の空白を削除 - trim(col) df = df.withColumn("name", trim(col("name"))); // 左側を埋める - lpad(col, len, pad) // 右側を埋める - rpad(col, len, pad) df = df.withColumn("id", lpad(col("id"), 4, "0")); // 左側の空白を削除 - ltrim(col) // 右側の空白を削除 - rtrim(col) df = df.withColumn("id", ltrim(col("id"))); // 連結 - concat(Column... cols) （いずれかの列が null の場合は null） df = df.withColumn("full_name", concat(col("fname"), lit(" "), col("lname"))); // 区切り文字を使って連結 - concat_ws(delim, Column... cols) （null を無視します） df = df.withColumn("full_name", concat_ws("-", "fname", "lname")); // 正規表現で置換 - regexp_replace(col, pattern, replacement) df = df.withColumn("id", regexp_replace(col("id"), "0F1(.*)", "1F1-$1")); // 正規表現で抽出 - regexp_extract(str, pattern, idx) df = df.withColumn("id", regexp_extract(col("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// 四捨五入 - round(col, scale=0) df = df.withColumn("price", round(col("price"), 0)); // 切り下げ - floor(col) df = df.withColumn("price", floor(col("price"))); // 切り上げ - ceil(col) df = df.withColumn("price", ceil(col("price"))); // 絶対値 - abs(col) df = df.withColumn("price", abs(col("price"))); // X の Y 乗 – pow(X, Y) df = df.withColumn("exponential_growth", pow(col("x"), 2.0)); // 複数の列から最小値を選択します – least(Column... cols) df = df.withColumn("least", least(col("subtotal"), col("total"))); // 複数の列から最大値を選択します – greatest(Column... cols) df = df.withColumn("greatest", greatest(col("subtotal"), col("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", to_date(col("date_of_birth"), "yyyy-MM-dd")); // 既知の形式の文字列をタイムスタンプに変換します（時刻情報を含みます） df = df.withColumn("time_of_birth", to_timestamp(col("time_of_birth"), "yyyy-MM-dd HH:mm:ss")); // 日付から年を取得します: year(col) // 日付から月を取得します: month(col) // 日付から日を取得します: dayofmonth(col) // 日付から時を取得します: hour(col) // 日付から分を取得します: minute(col) // 日付から秒を取得します: second(col) df = df.filter(year(col("date_of_birth")).equalTo("2017")); // 日数を加算・減算します df = df.withColumn("three_days_after", date_add(col("date_of_birth"), 3)); df = df.withColumn("three_days_before", date_sub(col("date_of_birth"), 3)); // 月数を加算・減算します df = df.withColumn("next_month", add_months(col("date_of_birth"), 1)); df = df.withColumn("previous_month", add_months(col("date_of_birth"), -1)); // 2つの日付の間の日数を取得します df = df.withColumn("days_between", datediff(col("end"), col("start"))); // 2つの日付の間の月数を取得します df = df.withColumn("months_between", months_between(col("end"), col("start"))); // date_of_birth が 2017-05-10 から 2018-07-21 の間にある行のみを保持します df = df.filter( (col("date_of_birth").geq("2017-05-10")).and( (col("date_of_birth").leq("2018-07-21"))) );
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14// 既存の列から配列または構造体の列を作成します df = df.withColumn("guardians", array(col("guardian_1"), col("guardian_2"))); df = df.withColumn("properties", struct(col("hair_color"), col("eye_color"))); // インデックスまたはキーで配列または構造体の列から抽出します（無効な場合は null） df = df.withColumn("primary_guardian", col("guardians").getItem(0)); df = df.withColumn("hair_color", col("properties").getItem("hair_color")); // 配列または構造体の列を複数の行に展開します df = df.select(col("child_name"), explode(col("guardians"))); df = df.select(col("child_name"), explode(col("properties"))); // 構造体の列を複数の列に展開します df = df.select(col("child_name"), col("properties.*"));
Copied!1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16// 行の件数: count(col), countDistinct(col) // グループ内の行の合計: sum(col) // グループ内の行の平均: mean(col) // グループ内の行の最大値: max(col) // グループ内の行の最小値: min(col) // グループ内の最初の行: first(col, ignoreNulls) df = df.groupBy(col("address")).agg( count(col("uuid")).alias("num_residents"), max(col("age")).alias("oldest_age"), first(col("city"), true).alias("city") ); // グループ内のすべての行をセットに収集します: collect_set(col) // グループ内のすべての行をリストに収集します: collect_list(col) df = df.groupBy(col("address")).agg(collect_set("name").alias("resident_names"));
Copied!1 2// 再パーティション化 – df.repartition(num_output_partitions) df = df.repartition(1);
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 27import org.apache.spark.sql.expressions.UserDefinedFunction; import org.apache.spark.sql.types.DecimalType; import static org.apache.spark.sql.functions.col; import static org.apache.spark.sql.functions.udf; import java.math.BigDecimal; /** * 次の例では、Spark のデータ型「Decimal」の2列を加算する java UDF を作成します。 * Spark は DecimalType を java.math.BigDecimal で表すため、このクラスを使用します。 * ほかの Spark のデータ型は、https://spark.apache.org/docs/3.0.0/sql-ref-datatypes.html で定義されているとおり、ほかの Java のデータ型で表されます。 */ public final class HighLevelAutoTransform { @Compute @Output("...") public Dataset<Row> myComputeFunction(@Input("...") Dataset<Row> df) { UserDefinedFunction addsUDF = udf((BigDecimal i, BigDecimal j) -> { if (i == null || j == null) { // null の場合は必ず処理します return null; } return i.add(j); }, new DecimalType()); // これは UDF の結果の Spark データ型です return df.withColumn("a_plus_b", addsUDF.apply(col("a"), col("b"))); } }