こちらで紹介したように、Delta Live Tables(DLT)ではUDF(User Defined Function:ユーザー定義関数)を活用することで、処理をモジュール化することができます。

前回紹介したのは、integerを引数として渡すシンプルなUDFでしたが、ケースによっては以下のようにテーブルの全てのカラムを渡して処理を行いたいというケースがあるかと思います。

Python

@dlt.table
def silver_data():
  return (
    spark.sql("SELECT *, makeColSum(*) AS sumResult FROM live.bronze_data")
  )

現状、私が調べた範囲ではこちらのUDFに配列を渡すことができないので、以降で説明する、ある意味泥臭いやり方を取る必要があります。ただ、列数が変動しても柔軟に処理を行えるというメリットがあります。

こちらのアプローチは、列のすべての値を文字列結合してUDFに渡して、UDF側で値に分離して処理を行うというものです。

呼び出し側の準備

まず、ダミーデータを準備します。

dlt-python-caller

@dlt.table
def bronze_data():
  return spark.sql("SELECT id AS id, rand() * 100 AS col1, rand() * 100 AS col2 FROM RANGE(10)")

こちらのデータの全カラムの値の総和を取るUDFを呼び出します。

dlt-python-caller

@dlt.table
def silver_data():
  return (
    spark.sql("SELECT *, makeColSum(CONCAT_WS('_', *)) AS sumResult FROM live.bronze_data")
  )

ここでの肝はCONCAT_WSです。セパレーターを指定して全カラムを文字列結合します。これをUDFmakeColSumに渡します。

呼び出されるUDFの準備

dlt-python-callee

from pyspark.sql.functions import *
from pyspark.sql.types import *
import math

def colSum(x) -> float:
    """
    指定された行の総和を取るUDF
    :param x: Pyspark/SQLのカラムを`_`で連結したもの
    :return: 総和
    """
    
    # separatorで分割
    str_list = x.split('_')
    # floatのlistに変換
    float_list = list(map(float, str_list))
    # 総和の計算
    sum_result = math.fsum(float_list)

    return sum_result

spark.udf.register("makeColSum", colSum, FloatType())

パイプラインの実行

上記のノートブックを用いてパイプラインを作成して実行します。2つノートブックをノートブックライブラリとして追加します。

パイプラインの設定(JSON)は以下のようになります。

JSON

{
    "id": "432c0f50-99be-45e3-8860-02a8d4061dfa",
    "clusters": [
        {
            "label": "default",
            "autoscale": {
                "min_workers": 1,
                "max_workers": 5
            }
        }
    ],
    "development": true,
    "continuous": false,
    "edition": "advanced",
    "photon": false,
    "libraries": [
        {
            "notebook": {
                "path": "/Users/takaaki.yayoi@databricks.com/20220610_dlt_udf_process_all_columns/dlt-python-caller"
            }
        },
        {
            "notebook": {
                "path": "/Users/takaaki.yayoi@databricks.com/20220610_dlt_udf_process_all_columns/dlt-python-callee"
            }
        }
    ],
    "name": "dlt_custom_udf",
    "storage": "dbfs:/user/takaaki.yayoi@databricks.com/dlt",
    "target": "dlt_takaaki_yayoi_databricks_com_db"
}

結果の確認

このように全カラムの総和がsumResultに格納されています。

今回は基本的な処理のみをカバーしました。列名含めた文字列として渡して、特定の列を除外するといった実装を行うことも可能です。

Databricks 無料トライアル

Databricks 無料トライアル