トランスフォーム

Python

入力データセットが空の場合にデータセットのビルドを中止する

入力データセットまたはソースデータセットが空の場合、トランスフォームの実行を防止または中止するにはどうすればよいですか？

このコードは、PySpark トランスフォームを使用して、入力 DataFrame を出力先の場所に書き込む前に、空かどうかを確認します。DataFrame が空の場合、トランザクションが中止され、空の DataFrame の書き込みが防止されます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from pyspark.sql import functions as F
from transforms.api import transform, Input, Output


@transform(
    out=Output("/Palantir/output_location/datasets/not_process_empty_files"),
    source_df=Input("/Palantir/input_location/sometimes_empty"),
)
def compute(source_df, out):
    source_df = source_df.dataframe()

    # 書き込む行が1行以上あるかを確認します
    if len(source_df.head(1)) == 0:
        # 書き込む行がないため、トランザクションを中止します
        out.abort()
    else:
        # 書き込むため、APPEND になります。
        out.write_dataframe(source_df)

• 投稿日：2024-03-26
• タグ：code authoring, code repositories, python, abort

空の DataFrame を生成する

PySpark トランスフォーム内でデータフレームを作成し、出力に書き込むにはどうすればよいですか？

このコードは、ダミーキーとタイムスタンプ列を含む1行だけの空の DataFrame を作成する関数を定義します。その後、トランスフォームデコレーターを使用して、DataFrame を出力データセットのパスに書き込みます。

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 pyspark.sql import types as T
from pyspark.sql import functions as F

# 現在のタイムスタンプを追加するだけの空のデータフレームを生成します
def get_empty_df(ctx):
    # DataFrame のスキーマを定義します
    schema = T.StructType([T.StructField("key", T.StringType(), True)])
    
    # ダミーキーを含む1行の DataFrame を作成します
    df = ctx.spark_session.createDataFrame([("dummy_key",)], schema)
    
    # 現在のタイムスタンプを持つ新しい列 'when' を追加します
    df = df.withColumn('when', F.current_timestamp())
    
    return df

@transform(
    out=Output("output_dataset_path")
)
def out_1(ctx, out):
    # 空のデータフレームを作成します
    df = get_empty_df(ctx)

    # データフレームを出力に書き込みます
    out.write_dataframe(df)

• 投稿日：2024-03-20
• タグ：code authoring, code repositories, python

音声コードを使用したエンティティ名のあいまいマッチング

PySpark で音声コードを使用してエンティティ名のあいまいマッチングを行うにはどうすればよいですか？

このコードは、PySpark を使用してエンティティ名をクレンジングし、音声コードを生成し、Jaro 類似度指標を使用してエンティティ名のあいまいマッチングを行います。2つのデータセット間で類似したエンティティ名を照合する際に役立ちます。

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
54
55
56
57
58
59
60
61
62
from pyspark.sql import functions as F
from pyspark.sql import types as T
from transforms.api import transform_df, Input, Output
import re
import jellyfish


def _add_phonetic_codes(df):
    # 名前の各部分の音声コードを生成します
    df = df.withColumn(
        "name_part", F.split("cleaned_name", " ")
    ).withColumn(
        "name_part", F.explode("name_part")
    ).withColumn(
        "phonetic_code", F.soundex("name_part")
    ).drop("name_part")
    return df


@transform_df(
    Output(),
    entities2=Input(),
    entities1=Input(),
)
def compute(sanctions, entities):

    # テキストをクレンジングする UDF を設定します
    def clean_text(text):
        cleaned_text = re.sub(r" +", " ", re.sub(r"[./-]+", "", text)).lower()
        return cleaned_text

    clean_text_udf = F.udf(clean_text, T.StringType())

    # エンティティ名をクレンジングします
    entities2 = entities2.withColumn("cleaned_name", clean_text_udf(F.col("name")))
    entities1 = entities1.withColumn("cleaned_name", clean_text_udf(F.col("entity_name")))

    # 音声コードを追加します
    entities2 = _add_phonetic_codes(entities2)
    entities1 = _add_phonetic_codes(entities1)

    # あいまい結合
    matched_entities = entities1.join(
        entities2, on=["phonetic_code"], how="inner"
    ).select(
        entities1.cleaned_name.alias("cleaned_name1"), entities1.id.alias("entity_id1")
        entities2.cleaned_name.alias("cleaned_name2"), entities2.id.alias("entity_id2")
    ).drop("phonetic_code")
    matched_entities = matched_entities.dropDuplicates()

    # 文字列を比較する UDF を設定します
    @F.udf()
    def jaro_compare(name1, name2):
        return jellyfish.jaro_similarity(name1, name2)

    # あいまいマッチング
    matched_entities = matched_entities.withColumn(
        "match_score", jaro_compare("cleaned_name1", "cleaned_name2")
    )
    matched_entities = matched_entities.filter(entities.match_score > 0.75)
    matched_entities = matched_entities.select("entity_id1", "entity_id2")
    return matched_entities

• 投稿日：2024-05-23
• タグ：pyspark, fuzzy matching, phonetic codes, jaro similarity

PySpark を使用して ORC ファイルを読み込む

PySpark を使用して ORC ファイルを読み込むにはどうすればよいですか？

このコードは、入力データセットの Hadoop パスから生の ORC ファイルを読み込み、生成された Spark データフレームを出力に書き込みます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
from transforms.api import transform, Input, Output


@transform(
    out=Output("output"),
    raw=Input("input"),
)
def compute(ctx, out, raw):
    hadoop_path = raw.filesystem().hadoop_path
    df = ctx.spark_session.read.format('orc').load(f'{hadoop_path}/')
    out.write_dataframe(df)

• 投稿日：2024-07-18
• タグ：pyspark, dataframe, orc, hadoop

複数の入力と出力を持つトランスフォーム

Foundry で複数の入力と出力を持つトランスフォームを作成するにはどうすればよいですか？

このコードは、複数の入力データセットを受け取り、複数の出力データセットを生成する PySpark トランスフォームの作成方法を示します。@transform デコレーターを使用し、入力と出力に明示的に名前を付けます。トランスフォームは入力データフレームを読み込んで処理し、結果を指定された出力データセットに書き込みます。

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

# @incremental デコレーターの有無（どちらにも対応）
# @transform_df を @transform に変更しました
# これにより、入力と出力をより細かく制御できます
# また、出力に明示的に名前を付ける必要があります
@transform(
    output_dataset_1=Output("3_multi_output_1"),
    output_dataset_2=Output("3_multi_output_2"),
    input_dataset_1=Input("fake_dataset"),
    input_dataset_2=Input("fake_dataset_2")
)
def example_transform_multi_inputs_outputs(input_dataset_1, input_dataset_2, output_dataset_1, output_dataset_2):
    # 入力を読み込めます
    input_df_1 = input_dataset_1.dataframe()
    input_df_2 = input_dataset_2.dataframe()

    # ここでの処理例
    only_data_from_1 = input_df_1.withColumn('processed_at', F.current_timestamp())
    unioned_version = input_df_1.unionByName(input_df_2).withColumn('processed_at', F.current_timestamp())

    # 変更したデータフレームを返すのではなく、"write_dataframe" を明示的に呼び出す必要があります。
    # 任意の出力に対して、書き込むデータフレームを指定して呼び出します。
    output_dataset_1.write_dataframe(only_data_from_1)
    output_dataset_2.write_dataframe(unioned_version)

• 投稿日：2024-03-20
• タグ：code authoring, code repositories, python