ロギング

Foundry の PySpark から、さまざまなデバッグ情報を出力できます。

Code Workbook

Python の組み込みの print は、Code Workbook のコードエディターの右側にある、通常エラーが表示される出力セクションに出力を送ります。

Copied!
1
2
def new_dataset(some_input_dataset):
    print("example log output")

example log output

Code Repositories

Code Repositories は、Python の組み込みの logging ライブラリ ↗ を使用します。このライブラリについてはオンラインに豊富なドキュメントがあり、ログレベル（ERROR、WARNING、INFO）を制御して、ログを簡単に絞り込めます。

ログ出力は、出力データセットのログファイルとビルドのドライバーログの両方に表示されます。データセットのログファイルはデータセット -> 詳細 -> ファイル -> ログファイル、ビルドのドライバーログはビルド -> ビルド -> ジョブステータスログでドライバーログを選択すると確認できます。

Copied!
1
2
3
4
5
6
7
8
9
import logging

logger = logging.getLogger(__name__)

@transform_df(
    ...
)
def some_transformation(some_input):
    logger.info("example log output")

INFO [2018-01-01T12:00:00] some_transformation: example log output

Python UDF 内からログを出力する

Spark は、上記の some_transformation 関数など、クエリを作成する最上位のドライバープロセスからのログ出力をキャプチャします。ただし、ユーザー定義関数（UDF）の内部で書き込まれたログはキャプチャしません。PySpark クエリ内で UDF を使用していてデータをログに記録する必要がある場合は、キャプチャしたいデータを返す2つ目の UDF を作成し、呼び出します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@transform_df(
    ...
)
def some_transformation(some_input):
    logger.info("log output related to the overall query")
    
    @F.udf("integer")
    def custom_function(integer_input):
    	return integer_input + 5
    
    @F.udf("string")
    def custom_log(integer_input):
    	return "Original integer was %d before adding 5" % integer_input
    
    df = (
    	some_input
    	.withColumn("new_integer", custom_function(F.col("example_integer_col"))
    	.withColumn("debugging", custom_log(F.col("example_integer_col"))
    )

例

クエリ内で何が起きているかについて、情報をログに記録したいことがよくあります。PySpark には DataFrame の内部情報を調べる方法がいくつかあり、この情報を上記のロギング機構に送信できます。

これらの例では Code Workbook の print 構文を使用しますが、Transforms & Authoring の logger の代わりに print を使用できます。

DataFrame の列

df.columns を使用すると、DataFrame に存在する列を調べられます。これにより、文字列のリストが生成されます。

Copied!
1
2
3
4
5
def employee_phone_numbers(employee_df, phone_number_df):
    print("employee columns are {}".format(employee_df.columns))
    print("phone columns are {}".format(phone_df.columns))
    df = employee_df.join(phone_number_df, 'employee_id', 'left')
    print("joined columns are {}".format(df.columns))

employee columns are ['name', 'employee_id']
phone columns are ['phone_number', 'employee_id']
joined columns are ['name', 'employee_id', 'phone_number']

結合の動作を確認する

左結合を実行し、一対一の関係を想定していて、左側の DataFrame の行数が変わっていないことを確認したいとします。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
def employee_phone_numbers(employee_df, phone_number_df):
    original_employee_rows = employee_df.count()
    print("Incoming employee rows {}".format(original_employee_rows))

    df = employee_df.join(phone_number_df, 'employee_id', 'left')
    rows_after_join = df.count()
    print("Final employee rows {}".format(rows_after_join))

    if rows_after_join > original_employee_rows:
        print("Some employees have multiple phone numbers!")
    else:
        print("Data is correct")

Incoming employee rows 100
Final employee rows 105
Some employees have multiple phone numbers!

Spark クエリプラン

.explain() を呼び出すと、特定の DataFrame を生成するために Spark が実行する、最適化された物理プランにアクセスできます。

Copied!
1
2
3
4
def employee_phone_numbers(employee, phone):
    employee = employee.where(F.month(employee.birthday) == F.month(F.current_date()))
    df = employee.join(phone, 'employee_id', 'left')
    df.explain()

== Physical Plan ==
*(2) Project [employee_id#9734, name#9732, birthday#9733, phone_number#9728]
+- *(2) BroadcastHashJoin [employee_id#9734], [employee_id#9729], LeftOuter, BuildRight
   :- *(2) Filter (month(birthday#9733) = 10)
   :  +- *(2) FileScan parquet !ri.foundry.main.transaction.00000000-e98a-c557-a20f-5eea5f373e36:ri.foundry.main.transaction.00000000-e98a-c557-a20f-5eea5f373e36@00000000-1ebd-4a81-9f64-2d4c8a8472bc:master.ri.foundry.main.dataset.6ad20cd7-45b0-4312-b096-05f57487f650[name#9732,birthday#9733,employee_id#9734] Batched: true, Format: Parquet, Location: FoundryCatalogFileIndex[sparkfoundry://.../datasets/ri.f..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<name:string,birthday:date,employee_id:int>
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[1, int, true] as bigint)))
      +- *(1) Project [phone_number#9728, employee_id#9729]
         +- *(1) Filter isnotnull(employee_id#9729)
            +- *(1) FileScan csv !ri.foundry.main.transaction.00000000-e989-4f9a-90d5-996f088611db:ri.foundry.main.transaction.00000000-e989-4f9a-90d5-996f088611db@00000000-1ebc-f483-b75d-dbcc3292d9e4:master.ri.foundry.main.dataset.f5bf4c77-37c0-4e29-8a68-814c35442bbd[phone_number#9728,employee_id#9729] Batched: false, Format: CSV, Location: FoundryCatalogFileIndex[sparkfoundry://.../datasets/ri.f..., PartitionFilters: [], PushedFilters: [IsNotNull(employee_id)], ReadSchema: struct<phone_number:int,employee_id:int>

データを表示する

電話番号を最も多く持つ従業員を確認したいとします。次の例では、対象のデータセット（電話番号を2つ以上持つ従業員）を作成し、.take(3) を呼び出して先頭の3行をリストとして取得します。また、.collect() を使用すると、DataFrame のすべての行をリストとして取得できます。

Copied!
1
2
3
4
5
6
def multiple_numbers(phone_numbers):
    df = phone_numbers.groupBy('employee_id').agg(
        F.count('phone_number').alias('numbers')
    ).where('numbers' > 1).sort(F.col('numbers').desc())

    print(df.take(3))

[Row(employee_id=70, numbers=4), Row(employee_id=90, numbers=2), Row(employee_id=25, numbers=2)]