Documentation
ドキュメントの検索
karat

+

K

APIリファレンス ↗
コードの例増分トランスフォームトランスフォーム

注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

トランスフォーム

Python

PySpark を使用して API からデータを取得し、インクリメンタルに更新する

外部トランスフォームを使用して API からデータを取得し、インクリメンタルに更新するにはどうすればよいですか？

このコードは PySpark と requests ライブラリを使用して、指定した日付範囲のデータを API から取得し、出力をインクリメンタルに更新します。API がページングをサポートしている場合は、ページングもサポートします。

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from pyspark.sql import functions as F
from transforms.api import incremental, transform, Output
import requests
from transforms.external.systems import EgressPolicy, use_external_systems, Credential
import logging
from datetime import datetime as dt
import json


def _get_data(token, start_date, end_date, next_link_url='<YOUR_URL>'):
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}

    data = {
        "from": start_date,
        "to": end_date,
    }
    response = requests.post(next_link_url, json=data, headers=headers)
    logging.warn(response.json())
    data = response.json()["data"]
    return json.dumps(data)


@use_external_systems(
    creds=Credential(),
    egress=EgressPolicy(),
)
@incremental()
@transform(
    output=Output(),
)
def compute(output, creds, egress, ctx):
    token = creds.get("token")
    if ctx.is_incremental:
        previous = output.dataframe('current').localCheckpoint()
        if NEXT_LINK_COLUMN in previous.columns:
            latest_row = (
                previous
                .where(F.col(LAST_MODIFIED_COLUMN).isNotNull())
                .orderBy([F.col(REQUEST_TIMESTAMP_COLUMN).desc(), F.col(LAST_MODIFIED_COLUMN).desc()])
                .limit(1).collect()[0]
            )
            next_link_url = latest_row[NEXT_LINK_COLUMN]
            last_date = latest_row[LAST_MODIFIED_COLUMN]
        else:
            last_date = previous.orderBy(F.col(LAST_MODIFIED_COLUMN).desc()).limit(1).collect()[0][LAST_MODIFIED_COLUMN]
    
    today = dt.today().strftime("%Y-%m-%d")
    
    data = _get_data(token, last_date, today, next_link_url)
    
    df = ctx.spark_session.createDataFrame([{'date': last_date, 'data': data}])
    output.set_mode("modify")
    output.write_dataframe(df)
  • 投稿日：2024-04-26
  • タグ：API、pyspark、incremental、dataframe、external transform

インクリメンタル（差分処理）ヘルパー

インクリメンタルトランスフォームがインクリメンタルに実行されているかどうかに応じて、入力を変更するにはどうすればよいですか？

Copied!
1
2
3
4
5
6
7
8
9
10
11
@incremental()
@transform(
	x=Output(),
	y=Input(),
	z=Input()
)
def compute(ctx, x, y, z):
	if ctx.is_incremental:
		## Some Code
	else:
		## Other Code
  • 投稿日：2024-03-20
  • タグ：code authoring、code repositories、python、incremental

インクリメンタルな合計集計

インクリメンタルデコレーターを使用して非インクリメンタルパイプラインを最適化し、非常に大きなデータセットでインクリメンタルな集計を行うにはどうすればよいですか？

このコードは PySpark を使用して、特定のフィールドに基づいてデータセットをグループ化し、別のフィールドの一意な値を数えることで、データセットの日次集計を計算します。次に、結果を出力データフレームに格納し、インクリメンタルと非インクリメンタルの両方のケースに対応します。

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
from pyspark.sql import functions as F
from transforms.api import transform, incremental, Input, Output
from pyspark.sql import DataFrame


@incremental(semantic_version=1)
@transform(
    # 入力データセット
    input_data=Input(""),
    daily_aggregate=Output("")
)
def compute(ctx, input_data, daily_aggregate):
    input_df = input_data.dataframe()
    latest_daily_agg = input_df.groupBy(F.col("group_by_field")).agg(F.count_distinct(F.col("unique_thing")).alias("sum_of_unique"))
    # 以前の出力を読み込むにはスキーマが必要です。
    latest_daily_agg_schema = latest_daily_agg.schema

    if ctx.is_incremental:

        last_daily_agg = daily_aggregate.dataframe(mode='previous', schema=latest_daily_agg_schema)
        sum_daily = last_daily_agg.unionByName(latest_daily_agg).groupBy(F.col("group_by_field")).agg(F.sum(F.col("sum_of_unique")).alias("sum_of_unique"))
        daily_aggregate.set_mode('replace')
        daily_aggregate.write_dataframe(sum_daily)

    else:
        # インクリメンタルではない場合は、最新の日次集計のみを保存します
        daily_aggregate.write_dataframe(latest_daily_agg)
  • 投稿日：2024-03-20
  • タグ：code authoring、code repositories、python、incremental、aggregation

PySpark のインクリメンタルトランスフォーム

PySpark を使用して Palantir Foundry でインクリメンタルトランスフォームを実装するにはどうすればよいですか？

このコードは、PySpark トランスフォームでインクリメンタルデコレーターを使用し、インクリメンタル処理に対応する方法を示します。入力データフレームに processed_at タイムスタンプ列と is_running_as_incremental 列を追加します。

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
from transforms.api import transform_df, Input, Output, incremental
from pyspark.sql import types as T
from pyspark.sql import functions as F

# インクリメンタル（差分処理）デコレーターを適用します
@incremental()
@transform_df(
    Output(""),
    input_df=Input("")
)
def example_transform_incremental_processing(ctx, input_df):
    # トランスフォームの動作は入力と出力の状態によって異なります
    # （詳細は元のコードスニペットのケース1から3を参照してください）
    #
    # 以下のロジックはスナップショット処理と変わりません。
    # 動作は Foundry によって制御されます。

    # ここで処理の例を示します
    input_df = input_df.withColumn('processed_at', F.current_timestamp())
    
    # ctx.is_incremental を使用して、Foundry が現在のビルドを「スナップショット」として実行しているか、「インクリメンタル」として実行しているかを確認します
    input_df = input_df.withColumn('is_running_as_incremental', F.lit(ctx.is_incremental))

    # 編集したデータフレームを返します
    return input_df
  • 投稿日：2024-03-20
  • タグ：code authoring、code repositories、python、incremental

スナップショット入力からインクリメンタル出力へ

完全に書き換えられるスナップショット入力に対してインクリメンタルトランスフォームを実行し、まだ処理していない新しい行だけを処理するにはどうすればよいですか？

このコードは PySpark を使用して入力データセットをスナップショットとして読み取り、前回の出力と比較して新しい行を見つけます。次に、新しい行にタイムスタンプを追加して出力データセットに追加することで、インクリメンタルな変換を実行します。

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
from transforms.api import incremental, transform, Input, Output
from pyspark.sql import types as T, functions as F

# @incremental デコレーターで高度な読み取り/書き込みモードを利用できます。
@incremental(
    snapshot_inputs=["input_dataset"]
)
# @transform で入力と出力をより細かく制御できます。
@transform(
    output_dataset=Output("incremental_output"),
    input_dataset=Input("snapshot_input")
)
def example_transform_very_advanced_processing__snapshot_to_incremental(ctx, input_dataset, output_dataset):
    # 入力データフレームを必ずスナップショットとして読み取ります
    input_df_all_dataframe = input_dataset.dataframe(mode="current")

    # 現在の出力を読み取り、以前のビルドですでに処理した内容を確認します
    out_schema = T.StructType([T.StructField("primary_key", T.StringType(), True),
                               T.StructField("other_column", T.IntegerType(), True), # 出力に保存するほかの列を含めます
                               T.StructField("processed_at", T.TimestampType(), True)])
    output_df_previous_dataframe = output_dataset.dataframe('previous', out_schema)

    # ==== 処理の例 ====
    # 入力と現在の出力の差分を取り、「新しい行」を見つけます。
    KEY = ["primary_key"]
    new_rows_df = input_df_all_dataframe.join(output_df_previous_dataframe, how="left_anti", on=KEY)
    # この例の追跡、デバッグ、理解を容易にするため、タイムスタンプを追加します
    new_rows_df = new_rows_df.withColumn('processed_at', F.current_timestamp())

    # ==== 処理の例の終わり ====

    # これにより行が追加されます
    output_dataset.set_mode("modify")
    output_dataset.write_dataframe(new_rows_df)
  • 投稿日：2024-07-18
  • タグ：code authoring、code repositories、python、incremental