トランスフォーム

Python

Python で Excel ファイルをパースする

PySpark と Openpyxl を使用して、動的なスキーマを持つ複雑な Excel ファイルを分散処理で読み込み、処理するにはどうすればよいですか？

このコードでは、PySpark と Openpyxl ライブラリを使用して、入力ファイルシステムから複数の Excel ファイルを読み込み、その内容をパースして PySpark DataFrame に変換します。その後、これらの DataFrame を1つの DataFrame に結合し、出力に書き込みます。

parse_workbook と convert_to_df はプレースホルダーのヘルパー関数であり、Foundry API の一部ではありません。関数の本体を、ワークブックの構造に合わせたロジックに置き換えてください。以下の実装例では、アクティブなシートの最初の行をヘッダーとして読み込み、残りの行から 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
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
from pyspark.sql import functions as F, types as T, DataFrame
from transforms.api import transform, Input, Output, configure
import tempfile
import shutil
import openpyxl
import functools


@transform(
    processed_excel=Output("example_processed_dataframe"),
    excel_input=Input("example_excel_dataframe"),
)
def compute(ctx, processed_excel, excel_input):
    def parse_file(file_status):
        # Excel ファイルを開きます
        with excel_input.filesystem().open(file_status.path, "rb") as in_xlsx:
            # 処理用の一時ファイルを作成します
            with tempfile.NamedTemporaryFile(suffix=".xlsx") as tmp_xlsx:
                shutil.copyfileobj(in_xlsx, tmp_xlsx)
                tmp_xlsx.flush()

                # Excel ワークブックを読み込み、内容をパースします
                try:
                    workbook = openpyxl.load_workbook(tmp_xlsx.name)
                    return parse_workbook(workbook)
                except:
                    return None

    # 入力ファイルシステムから Excel ファイルの一覧を取得します
    files_df = excel_input.filesystem().files()
    # parse_file 関数を使って各ファイルをパースします
    parsed_files = files_df.rdd.map(parse_file).collect()

    # パースしたファイルを PySpark データフレームに変換します
    dfs = []
    for parsed_file in parsed_files:
        dfs.append(convert_to_df(ctx, parsed_file))

    # データフレームの和集合を作成し、結果を出力に書き込みます
    df = functools.reduce(lambda df1, df2: df1.unionByName(df2, allowMissingColumns=True), dfs)
    processed_excel.write_dataframe(df)


# プレースホルダー：ワークブックの内容を通常の Python オブジェクトに抽出します。
# 関数本体をワークブックの構造に合わせたロジックに置き換えます。
def parse_workbook(workbook):
    # 例：アクティブなシートの1行目をヘッダーとして使用し、
    # 残りの行を辞書のリストとして返します
    sheet = workbook.active
    rows = list(sheet.iter_rows(values_only=True))
    if not rows:
        return []
    header = [str(cell) for cell in rows[0]]
    return [dict(zip(header, row)) for row in rows[1:]]


# プレースホルダー：パースした Python オブジェクトを PySpark DataFrame に変換します。
# parse_workbook が返す構造に合わせて関数本体を置き換えます。
def convert_to_df(ctx, parsed_file):
    # 例：辞書のリストから DataFrame を作成します
    return ctx.spark_session.createDataFrame(parsed_file)

• 投稿日：2024-08-12
• タグ：code authoring, code repositories, python, openpyxl

シェープファイルを結合して GeoJSON に変換する

複数のシェープファイルを結合して GeoJSON 形式に変換するにはどうすればよいですか？

このコードでは、geospatial_tools ライブラリを使用して複数のシェープファイルを読み込み、ジオメトリを GeoJSON 形式に変換して、1つの PySpark 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from transforms.api import transform, Input, Output
from geospatial_tools import geospatial
from geospatial_tools.functions import clean_geometry, centroid, geojson_to_geohash
import tempfile
import shutil
import geopandas as gpd
from pyspark.sql import types as T
from pyspark.sql import functions as F
import json
from shapely.geometry import mapping


@geospatial()
@transform(
    output=Output(),
    input_data=Input(),
)
def compute(ctx, input_data, output):
    fs = input_data.filesystem()
    schema = T.StructType([T.StructField("geoshape", T.StringType()),
                           T.StructField("name", T.StringType()),
                           T.StructField("centroid", T.StringType())])
    shapefiles = [f.path.replace('.shp', '') for f in fs.ls(glob='*shp')]
    combined_data = ctx.spark_session.createDataFrame([], schema)
    for shapefile in shapefiles:  # NOQA
        with tempfile.TemporaryDirectory() as tmp_dir:
            # シェープファイルに関連するすべてのファイルをローカルファイルシステムにコピーします
            # シェープファイルには、.prj や .cpg などの複数のファイルが関連付けられています
            for shapefile_file in fs.ls(glob=f'{shapefile}.*'):
                with open(f'{tmp_dir}/{shapefile_file.path}', 'wb') as tmp_file:
                    with fs.open(shapefile_file.path, 'rb') as f:
                        shutil.copyfileobj(f, tmp_file)
            # GeoJSON ジオメトリ列を作成します
            pdf = gpd.read_file(f'{tmp_dir}/{shapefile}.shp')
            pdf['geometry'] = pdf.geometry.apply(lambda x: json.dumps(mapping(x)))
            df = ctx.spark_session.createDataFrame(pdf)
            
            # すべてを Foundry が期待する EPSG:4326 形式に変換します
            crs = gpd.read_file(f'{tmp_dir}/{shapefile}.shp').crs.to_string()
            df = df.withColumn(
                "geoshape",
                clean_geometry('geometry', crs, lat_long=(crs != "EPSG:4326"))
                ).select("geoshape")
            df = df.withColumn('name', F.lit(shapefile))
            df = df.withColumn('centroid', geojson_to_geohash(centroid('geoshape')))
            combined_data = combined_data.unionByName(df)

    return output.write_dataframe(combined_data)

• 投稿日：2024-05-23
• タグ：geospatial, shapefile, geojson, geohash, pyspark, geopandas

データセット間で生のファイルをコピーする

Python で開発するトランスフォーム内で、入力データセットから出力データセットに生のファイルをコピーするにはどうすればよいですか？

このコードでは、入力データセットから出力データセットに生のファイルをコピーする PySpark トランスフォーム関数を定義します。「shutil」ライブラリを使用してファイルのバイトをコピーし、すべてのファイル、または指定された正規表現パターンに基づく一部のファイルのみをコピーできます。

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
from transforms.api import transform, Input, Output
from pyspark.sql import DataFrame
from functools import reduce
import shutil

# Python トランスフォームで入力データセットから出力データセットに生ファイルをコピーします
@transform(
    my_output=Output("my_output_dataset"),
    my_input=Input("my_input_dataset")
)
def copy_my_input(my_output, my_input):
    copy_raw_files(my_output, my_input, [".*\.csv"], False)


def copy_raw_files(my_output, my_input, regexes, copy_full=False):
    # 生ファイルをコピーします
    def copy_file(file_status):
        # 入力データフレームのファイルシステムで指定されたファイルを開きます
        with my_input.filesystem().open(file_status.path, 'rb') as in_f:
            # 出力データフレームのファイルシステムでファイルを開きます
            with my_output.filesystem().open(file_status.path, 'wb') as out_f:
                # ファイルのバイトを入力から出力にコピーします
                shutil.copyfileobj(in_f, out_f)

    # すべてのファイルをコピーするか、一部のみをコピーするかを選択します
    if copy_full:
        files_df = my_input.filesystem().files()
    else:
        files_to_copy = []
        for regex in regexes:
            # 正規表現に一致するファイルのみをコピーします
            files_to_copy.append(my_input.filesystem().files(regex=regex))
        # すべてのファイルを含む1個のデータフレームを作成します
        files_df = reduce(DataFrame.unionByName, files_to_copy)

    # これによりコピー操作が並列化されます
    files_df.rdd.foreach(copy_file)

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

ファイル処理

PySpark を使用してデータセット内の複数のファイルを処理するにはどうすればよいですか？

このコードでは、PySpark を使用して、gzip 圧縮されたファイルを含むデータセット内の複数のファイルを処理します。各ファイルの最初の行を読み込み、ファイル情報と最初の行の内容を含むデータフレームを作成します。

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
63
64
65
66
67
68
69
70
71
72
73
from transforms.api import transform, Input, Output, incremental
from pyspark.sql import types as T
from pyspark.sql import functions as F
from pyspark.sql import Row
import gzip
import io

# @incremental デコレーターの有無（どちらにも対応）
# @transform_df を @transform に変更しました
# これにより入力と出力をより細かく制御でき、入力データセットの「ファイル」バージョンにアクセスするために必要です


@transform(
    output_dataset_1=Output(""),
    output_dataset_2=Output(""),
    input_dataset=Input("")
)
def example_transform_file_processing(ctx, input_dataset, output_dataset_1, output_dataset_2):
    # files() メソッドは入力データセットのファイルシステムを表すデータフレームを返します
    fs = input_dataset.filesystem()
    files_df = input_dataset.filesystem().files()

    # ここで各ファイルのパスを抽出し、必要な方法や任意の方法で処理できます
    # ==== 計算の例
    # rdd flatmap の出力スキーマを定義します
    schema = T.StructType([
        T.StructField('hadoop_path', T.StringType()),
        T.StructField('file_name', T.StringType()),
        T.StructField('size', T.LongType()),
        T.StructField('modified', T.LongType()),
        T.StructField('first_row_content', T.StringType())
    ])
    cols = schema.fieldNames()  # 次と同等です :["hadoop_path", "file_name", "size", "modified"]
    MyRow = Row(*cols)  # RDD の UDF のような関数の戻り値の型として使用する MyRow オブジェクトを定義します

    # 1個のファイルをパースするインライン関数（考え方：UDF に似ていますが、RDD 用です）
    def process_file(file_status):
        # 処理の例：各ファイルの1行目を読み込みます
        line = "default value"
        try:
            line = "WARNING: Not supported file type."
            if file_status.path.endswith('.gz'):
                # Gzip 圧縮されたファイルを処理します
                with fs.open(file_status.path, "rb") as f:
                    gz = gzip.GzipFile(fileobj=f)
                    br = io.BufferedReader(gz)
                    tw = io.TextIOWrapper(br)
                    line = tw.readline()
            else:
                with fs.open(file_status.path, "r") as f:
                    line = f.readline()

        except Exception as e:
            line = "ERROR: " + str(e)

        # RDD の要素から行を作成します
        yield MyRow(fs.hadoop_path, file_status.path, file_status.size, file_status.modified, line)

    # ファイルのデータフレームを RDD に変換します。https://spark.apache.org/docs/latest/rdd-programming-guide.html を参照してください
    rdd = files_df.rdd
    # RDD の各要素に関数を適用します
    rdd = rdd.flatMap(process_file)
    # 出力への書き込みを容易にするため、RDD をデータフレームに変換します
    # スキーマを指定すると、空の rdd を処理できます。
    output_df = ctx.spark_session.createDataFrame(rdd, schema)
    # タイムスタンプを追加します
    output_df = output_df.withColumn('processed_at', F.current_timestamp())
    # ==== 計算の例の終わり

    # 入力のファイルシステムを表すデータフレームを出力に書き込みます
    output_dataset_1.write_dataframe(files_df)
    # 処理したデータフレームを出力に書き込みます
    output_dataset_2.write_dataframe(output_df)

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

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

PySpark を使用して SAS ファイルをパースする

SAS データセットから PySpark データフレームを作成するにはどうすればよいですか？

このコードでは、生の SAS ファイルを含む入力データセットを受け取り、それらのファイルから PySpark データフレームを作成するトランスフォーム関数を定義します。spark-sas7bdat パッケージを使用して SAS ファイルを読み込み、データフレームに読み込みます。

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
@transform(
    output=Output("xxxxx"), # ここに Foundry RID を指定します
    input_df=Input("xxxxx") # ここに Foundry RID を指定します
)
def parse_sas_file(ctx, input_df, output, sas_path="*.sas7bdat"):
    '''
    SAS データセットから PySpark データフレームを作成します
    この関数はドライバーで計算を実行するため、ドライバーのメモリーを増やす必要がある場合があります
    ctx: Spark コンテキスト
    input_df: 生の SAS ファイルを含む入力データセット
    sas_path: データセット内の SAS ファイルへのパス。デフォルトはデータセット内のすべての SAS ファイルです
    include_filename_as_field: 下流でパースするため、ファイル名を列として含めます。デフォルトは false です
    '''
    fs = input_df.filesystem()
    hadoop_path = fs.hadoop_path
    files_df = fs.files(sas_path)
    # dfs = []

    spark_session = ctx.spark_session.builder.appName(ctx.spark_session.sparkContext.appName).config('spark.jars.packages', 'saurfang:spark-sas7bdat:3.0.0-s_2.12').getOrCreate()

    # TODO: 複数のパスに対応するように更新します
    # 元データセットから任意のファイルを読み込みます
    path = files_df.collect()[0].path
    full_path = f'{hadoop_path}/{path}'
    df = spark_session.read.format("com.github.saurfang.sas.spark").load(full_path)

    output.write_dataframe(df)

• 投稿日：2024-07-29
• タグ：pyspark, dataframe, sas, code repositories

複数のファイルを処理して結合する

データセット内の複数のファイルを処理し、1つの PySpark DataFrame に結合するにはどうすればよいですか？

このコードでは、複数のファイルを含む入力データセットを受け取り、各ファイルを個別に処理して、結果を1つの PySpark DataFrame に結合する PySpark トランスフォームを定義します。「map」関数を使用してデータセット内の各ファイルに「parse_file」関数を適用し、結果を収集して、すべての DataFrame の和集合を作成します。

process_file_locally_and_return_python_object と convert_to_df はプレースホルダーのヘルパー関数であり、Foundry API の一部ではありません。関数の本体を、ファイル形式に合わせたロジックに置き換えてください。以下の実装例では、改行区切りの JSON を読み込み、レコードのリストから 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
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
from pyspark.sql import functions as F, types as T, DataFrame
from transforms.api import transform, Input, Output, configure
import tempfile
import shutil
import functools
import json

# PySpark トランスフォームを定義します
@transform(
    processed_files=Output("example_processed_files_dataset"),
    file_dataset=Input("example_file_dataset"),
)
def compute(ctx, processed_files, file_dataset):
    # 1個のファイルをパースする関数
    def parse_file(file_status):
        with file_dataset.filesystem().open(file_status.path, "rb") as in_file:
            with tempfile.NamedTemporaryFile() as tmp_file:
                shutil.copyfileobj(in_file, tmp_file)
                tmp_file.flush()

                # ファイルをローカルで処理し、Python オブジェクトを返します。
                # これは独自のパースロジック用のプレースホルダーです（以下を参照）。
                return process_file_locally_and_return_python_object(tmp_file)

    # データセット内のファイルの一覧を取得します
    files_df = file_dataset.filesystem().files()
    
    # 各ファイルをパースし、結果を収集します
    parsed_files = files_df.rdd.map(parse_file).collect()

    dfs = []
    for parsed_file in parsed_files:
        # パースしたファイルを PySpark DataFrame に変換します
        dfs.append(convert_to_df(ctx, parsed_file))

    # すべての DataFrame の和集合を作成します
    df = functools.reduce(lambda df1, df2: df1.unionByName(df2, allowMissingColumns=True), dfs)
    
    # 結果の DataFrame を出力データセットに書き込みます
    processed_files.write_dataframe(df)


# プレースホルダー：ローカルの一時ファイルを読み込み、通常の Python オブジェクトを返します。
# 関数本体をファイル形式に合わせたロジックに置き換えます。
def process_file_locally_and_return_python_object(tmp_file):
    # 例：改行区切りの JSON レコードを辞書のリストに読み込みます
    with open(tmp_file.name, "r") as f:
        return [json.loads(line) for line in f if line.strip()]


# プレースホルダー：パースした Python オブジェクトを PySpark DataFrame に変換します。
# process_file_locally_and_return_python_object が返す構造に合わせて関数本体を置き換えます。
def convert_to_df(ctx, parsed_file):
    # 例：辞書のリストから DataFrame を作成します
    return ctx.spark_session.createDataFrame(parsed_file)

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

DOCX ファイルから内容を抽出する

Python を使用して DOCX ファイルから内容を抽出するにはどうすればよいですか？

このコードでは、python-docx ライブラリを使用してデータセットから DOCX ファイルの内容を読み込み、後続の処理のために Document オブジェクトに格納します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from transforms.api import transform, Input, Output
import docx as dx
from io import BytesIO

@transform(
    output=Output("output_dataset"),
    docs=Input("input_dataset"),
)
def compute(ctx, docs, output):
        
    fs = docs.filesystem()
    doc_file = list(fs.ls(regex=r'.*\.docx'))[0]

    # ファイルシステムを使ってファイルを開き、内容を BytesIO オブジェクトに読み込みます
    with fs.open(doc_file.path, 'rb') as f:
        source_stream = BytesIO(f.read())
        document = dx.Document(source_stream)
        source_stream.close()
        
    # document オブジェクトに対して何らかの処理を行います

• 投稿日：2024-03-20
• タグ：code authoring, python, python-docx, bytesio, raw files, unstructured

ZIP 圧縮された CSV を処理する

PySpark で、複数の CSV ファイルを含む ZIP ファイルを入力データセットから読み込んで処理し、処理したデータを出力データセットに書き込むにはどうすればよいですか？

このコードでは、PySpark を使用して入力データセット内の CSV を含む ZIP ファイルを読み込んで処理し、各 CSV の最初の行をスキップして、処理したデータを出力データセットに書き込みます。

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
from pyspark.sql import functions as F, types as T, DataFrame
from transforms.api import transform, Input, Output, configure
import shutil
import tempfile
import zipfile
import io

@transform(
    my_output=Output("my_output_dataset"),
    my_input=Input("my_input_dataset")
)
def compute(ctx, my_output, my_input):
    # 入力データセット内の各ファイルを処理する関数
    def process_file(file_status):
        with fs.open(file_status.path, 'rb') as f:
            with tempfile.NamedTemporaryFile() as tmp:
                shutil.copyfileobj(f, tmp)
                tmp.flush()

                # zip ファイルを読み込んで処理します
                with zipfile.ZipFile(tmp) as archive:
                    for filename in archive.namelist():
                        with archive.open(filename) as f2:
                            br = io.BufferedReader(f2)
                            tw = io.TextIOWrapper(br)
                            tw.readline() # 各 CSV の1行目をスキップします
                            # CSV の各行を読み込んで処理します
                            for line in tw:
                                yield MyRow(*line.split(","))
    
    # 入力データセットを読み込み、各ファイルを処理します
    rdd = my_input.files().rdd
    rdd = rdd.flatMap(process_file)
    df = rdd.toDF()

    # 処理したデータを出力データセットに書き込みます
    my_output.write_dataframe(df)

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

データセット内のファイルを解凍して抽出する

データセット内のファイルを解凍するにはどうすればよいですか？

このコードでは、PySpark を使用して入力から ZIP 圧縮されたファイルを読み込み、内容を抽出して、抽出したファイルを出力に書き込みます。ZIP 圧縮されたファイルを順に処理し、内容を BytesIO ストリームに読み込んでから、zipfile ライブラリを使用して一時ディレクトリにファイルを抽出します。その後、抽出したファイルを出力に書き込みます。

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
# from pyspark.sql import functions as F
from transforms.api import transform, Input, Output
import zipfile
import tempfile
import os
from io import BytesIO


@transform(
    unzipped=Output(""),
    zipped=Input(""),
)
def compute(unzipped, zipped):
    zip_files = zipped.filesystem().files(glob="*.zip").collect()
    for zip_file in zip_files:
        with zipped.filesystem().open(zip_file["path"], 'rb') as zip_f:
            source_stream = BytesIO(zip_f.read())
            with zipfile.ZipFile(source_stream, 'r') as zip_ref:
                with tempfile.TemporaryDirectory() as temp_dir:
                    zip_ref.extractall(temp_dir)
                    for path in iterate_directories(temp_dir):
                        output_file_name = path.replace(temp_dir, "")
                        with unzipped.filesystem().open(output_file_name, "w") as out_f:
                            with open(path, 'r') as in_f:
                                out_f.write(in_f.read())


def iterate_directories(directory):
    for root, dirs, files in os.walk(directory):
        for file in files:
            path = os.path.join(root, file)
            if is_leaf_file(path):
                yield path


def is_leaf_file(path):
    return os.path.isfile(path) and not os.path.islink(path)

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

データセットのファイルを ZIP 圧縮する

ファイルのデータセットから ZIP ファイルを作成するにはどうすればよいですか？

このコードでは、transforms API を使用してソースデータセットからすべての Markdown ファイルを読み込み、それらのファイルを含む ZIP ファイルを作成します。

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


@transform(
    my_output=Output(""),
    source_df=Input(""),
)
def compute(ctx, my_output, source_df):
    files = source_df.filesystem().files(glob="*.md").collect()

    with my_output.filesystem().open("foundry_code_examples.zip", 'wb') as write_zip:
        with zipfile.ZipFile(write_zip.name, 'w') as zip_file:
            for file_row in files:
                with source_df.filesystem().open(file_row["path"], 'rb') as markdown_file:
                    zip_file.write(markdown_file.name, arcname=file_row["path"])

    return source_df

• 投稿日：2024-03-26
• タグ：raw files, zip, python, code authoring, code repositories, export

Java

複雑な複数行のヘッダーを持つ Excel ファイルをパースする

ヘッダーが複雑で複数行からなる Excel ファイルをパースするにはどうすればよいですか？

このコードでは、transforms-excel-parser ライブラリを使用して、複雑なヘッダーを持つ Excel ファイルをパースする方法を示します。MultilayerMergedHeaderExtractor を指定して TableParser を作成し、その TableParser を指定して TransformsExcelParser を作成します。最後に、TransformsExcelParser を使用して入力データセット内の Excel ファイルからデータを抽出し、結果を出力に書き込みます。

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
package myproject.datasets;

import com.palantir.transforms.excel.ParseResult;
import com.palantir.transforms.excel.Parser;
import com.palantir.transforms.excel.TransformsExcelParser;
import com.palantir.transforms.excel.table.MultilayerMergedHeaderExtractor;
import com.palantir.transforms.excel.table.TableParser;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import java.util.Optional;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class ComplexHeaderExcel {

    @Compute
    public void myComputeFunction(
            @Input("<input_dataset_rid>") FoundryInput myInput,
            @Output("<output_dataset_rid>") FoundryOutput myOutput,
            @Output("<error_output_dataset_rid>") FoundryOutput errorOutput
        ) {
        // MultilayerMergedHeaderExtractor を使って TableParser を作成します
        Parser tableParser = TableParser.builder()
                .headerExtractor(MultilayerMergedHeaderExtractor.builder()
                        .topLeftCellName("A1")
                        .bottomRightCellName("D2")
                        .build())
                .build();

        // TableParser を使って TransformsExcelParser を作成します
        TransformsExcelParser transformsParser = TransformsExcelParser.of(tableParser);

        // 入力をパースします
        ParseResult result =
                transformsParser.parse(myInput.asFiles().getFileSystem().filesAsDataset());

        // パースしたデータを取得します。入力に行がなかった場合や
        // エラーが発生した場合は、空になることがあります
        Optional<Dataset<Row>> maybeDf = result.singleResult();

        // パースしたデータが空でなければ、出力データセットに書き込みます
        maybeDf.ifPresent(df -> myOutput.getDataFrameWriter(df).write());

        // エラー情報をエラー出力に書き込みます
        errorOutput.getDataFrameWriter(result.errorDataframe()).write();
    }
}

• 投稿日：2024-08-08
• タグ：code authoring, code repositories, java, transforms-excel-parser, excel

表形式ではない（フォーム）データを含む Excel ファイルをパースする

データが表形式ではない Excel ファイルをパースするにはどうすればよいですか？

このコードでは、transforms-excel-parser ライブラリを使用して、複数のシートにわたるフォームを含む Excel ファイルからデータを抽出する方法を示します。

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
63
64
65
66
67
68
69
70
71
72
package myproject.datasets;

import com.palantir.transforms.excel.TransformsExcelParser;
import com.palantir.transforms.excel.ParseResult;
import com.palantir.transforms.excel.Parser;
import com.palantir.transforms.excel.form.FieldSpec;
import com.palantir.transforms.excel.form.FormParser;
import com.palantir.transforms.excel.form.Location;
import com.palantir.transforms.excel.form.cellvalue.AdjacentCellAssertion;
import com.palantir.transforms.excel.form.cellvalue.CellValue;
import com.palantir.transforms.excel.functions.RegexSubstringMatchingSheetSelector;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;

public final class FormStyleExcel {
    private static final String FORM_A_KEY = "FORM_A";
    private static final String FORM_B_KEY = "FORM_B";

    @Compute
    public void myComputeFunction(
            @Input("<input_dataset_rid") FoundryInput myInput,
            @Output("<form_a_output_dataset_rid>") FoundryOutput formAOutput,
            @Output("<form_b_output_dataset_rid>") FoundryOutput formBOutput,
            @Output("<error_output_dataset_rid>") FoundryOutput errorOutput) {
        // フォーム A のパーサー設定
        Parser formAParser = FormParser.builder()
                .sheetSelector(new RegexSubstringMatchingSheetSelector("Form_A"))
                .addFieldSpecs(createFieldSpec("form_a_field_1", "B1"))
                .addFieldSpecs(createFieldSpec("form_a_field_2", "B2"))
                .build();

        // フォーム B のパーサー設定
        Parser formBParser = FormParser.builder()
                .sheetSelector(new RegexSubstringMatchingSheetSelector("Form_B"))
                .sheetSelector(new RegexSubstringMatchingSheetSelector("Form_B"))
                .addFieldSpecs(createFieldSpec("form_b_field_1", "B1"))
                .addFieldSpecs(createFieldSpec("form_b_field_2", "B2"))
                .build();

        // フォーム A とフォーム B の両方のパーサーを持つ TransformsExcelParser
        TransformsExcelParser transformsParser = TransformsExcelParser.builder()
                .putKeyToParser(FORM_A_KEY, formAParser)
                .putKeyToParser(FORM_B_KEY, formBParser)
                .build();

        // 入力をパースします
        ParseResult result =
                transformsParser.parse(myInput.asFiles().getFileSystem().filesAsDataset());

        // パースしたデータを出力データセットに書き込みます
        result.dataframeForKey(FORM_A_KEY)
                .ifPresent(df -> formAOutput.getDataFrameWriter(df).write());
        result.dataframeForKey(FORM_B_KEY)
                .ifPresent(df -> formBOutput.getDataFrameWriter(df).write());

        // エラー情報をエラー出力に書き込みます
        errorOutput.getDataFrameWriter(result.errorDataframe()).write();
    }

    // 適切なアサーションを持つ FieldSpec を簡潔に作成するヘルパーメソッド
    private static FieldSpec createFieldSpec(String fieldName, String cellLocation) {
        return FieldSpec.of(
                fieldName,
                CellValue.builder()
                        .addAssertions(AdjacentCellAssertion.left(1, fieldName))
                        .location(Location.of(cellLocation))
                        .build());
    }
}

• 投稿日：2024-08-06
• タグ：code authoring, code repositories, java, transforms-excel-parser, excel

単純な表形式の Excel ファイルをパースする

Transforms Excel Parser を使用して、単純な表形式の Excel ファイルをパースするにはどうすればよいですか？

このコードでは、transforms-excel-parser ライブラリを使用して、単純な表形式の Excel ファイルを含むデータセットをパースする方法を示します。SimpleHeaderExtractor を指定して TableParser を作成し、その TableParser を指定して TransformsExcelParser を作成します。最後に、TransformsExcelParser を使用して入力データセット内のファイルをパースし、抽出したデータを出力データセットに書き込みます。

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
package myproject.datasets;

import com.palantir.transforms.excel.ParseResult;
import com.palantir.transforms.excel.Parser;
import com.palantir.transforms.excel.TransformsExcelParser;
import com.palantir.transforms.excel.table.SimpleHeaderExtractor;
import com.palantir.transforms.excel.table.TableParser;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import java.util.Optional;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class SimpleTabularExcel {

    @Compute
    public void myComputeFunction(
            @Input("<input_dataset_rid>") FoundryInput myInput,
            @Output("<output_dataset_rid>") FoundryOutput myOutput,
            @Output("<error_output_dataset_rid>") FoundryOutput errorOutput
        ) {
        // 適切に設定した SimpleHeaderExtractor を使って TableParser を作成します
        // この例では、ファイルのヘッダーは2行目にあります。
        // ヘッダーが1行目にある場合は、
        // デフォルト値が0なので rowsToSkip を指定する必要はなく、実際には
        // TableParser.builder().build() だけで済みます。
        Parser tableParser = TableParser.builder()
                .headerExtractor(
                        SimpleHeaderExtractor.builder().rowsToSkip(1).build())
                .build();

        // TableParser を使って TransformsExcelParser を作成します
        TransformsExcelParser transformsParser = TransformsExcelParser.of(tableParser);

        // 入力をパースします
        ParseResult result =
                transformsParser.parse(myInput.asFiles().getFileSystem().filesAsDataset());

        // パースしたデータを取得します。入力に行がなかった場合や
        // エラーが発生した場合は、空になることがあります
        Optional<Dataset<Row>> maybeDf = result.singleResult();

        // パースしたデータが空でなければ、出力データセットに書き込みます
        maybeDf.ifPresent(df -> myOutput.getDataFrameWriter(df).write());

        // エラー情報をエラー出力に書き込みます
        errorOutput.getDataFrameWriter(result.errorDataframe()).write();
}

• 投稿日：2024-08-08
• タグ：code authoring, code repositories, java, transforms-excel-parser, excel