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

非構造化ファイルを読み書きする

データ変換でファイルにアクセスする理由はさまざまです。ファイルアクセスは、非テーブル形式（XML や JSON など）や圧縮形式（gz や zip など）のファイルを処理する場合に特に役立ちます。非構造化ファイルへのアクセスは高度なトピックであるため、このページを読む前に、このユーザーガイドのほかの内容を理解しておいてください。

transforms Python ライブラリを使用すると、Foundry データセット内のファイルを読み書きできます。transforms.api.TransformInput は読み取り専用の FileSystem オブジェクトを公開し、transforms.api.TransformOutput は書き込み専用の FileSystem オブジェクトを公開します。これらの FileSystem オブジェクトを使用すると、基盤となるストレージを意識することなく、Foundry データセット内のファイルのパスに基づいてファイルにアクセスできます。

データ変換でファイルにアクセスするには、transform() デコレーターを使用して Transform オブジェクトを構築する必要があります。これは、FileSystem オブジェクトが TransformInput オブジェクトと TransformOutput オブジェクトによって公開されるためです。計算関数への入力が TransformInput 型で、出力が TransformOutput 型であることを想定するデコレーターは、transform() だけです。

ファイルをインポートする

ファイルは、手動でインポートして Foundry にアップロードすることも、データ接続を介して同期することもできます。構造化ファイルと非構造化ファイルを Foundry データセットにインポートし、下流のアプリケーションで処理できます。また、拡張子を変更せずに、生のファイルとしてアップロードすることもできます。以下の例では、生のファイルとしてではなく、Foundry データセットとしてアップロードされたファイルを扱います。

Foundry には、データセットにアップロードされた特定の種類のファイルについて、スキーマを自動的に推論する機能もあります。たとえば、CSV 形式のファイルをインポートする際には、スキーマを自動的に適用するためのスキーマを適用ボタンを利用できます。データの手動アップロードについては、詳しくはこちらをご覧ください。

ファイルの一覧を見る

filesystem().ls() メソッドは、スキーマの有無にかかわらずデータセットで利用できます。ただし、.path()、.pandas()、.polars()、.arrow()、.filesystem().files() メソッドは、スキーマのあるデータセットでのみ利用できます。

データセット内のファイルは、transforms.api.FileSystem.ls() メソッドを使用して一覧表示できます。このメソッドは、transforms.api.FileStatus オブジェクトのジェネレーターを返します。これらのオブジェクトには、各ファイルのパス、サイズ（バイト単位）、更新タイムスタンプ（Unix エポックからのミリ秒数）が含まれます。次の Transform オブジェクトを例に説明します。

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(ctx, hair_eye_color, processed):
    # データ変換コード
    pass

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

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

@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

データ変換コード内で、データセットのファイルの一覧を見ることができます。

Copied!1
2
list(hair_eye_color.filesystem().ls())
# 結果: [FileStatus(path='students.csv', size=688, modified=...)]

glob または正規表現のパターンを渡して、ls() 呼び出しの結果を絞り込むこともできます。

Copied!1
2
3
4
5
list(hair_eye_color.filesystem().ls(glob='*.csv'))
# 結果: [FileStatus(path='students.csv', size=688, modified=...)]

list(hair_eye_color.filesystem().ls(regex='[A-Z]*\.csv'))
# 結果: []

ファイルを読み取る

ファイルは、transforms.api.FileSystem.open() メソッドを使用して開くことができます。このメソッドは、Python のファイルライクなストリームオブジェクトを返します。io.open() ↗ が受け付けるすべてのオプションにも対応しています。返されるストリームは seek メソッドと tell メソッドに対応していないため、ランダムアクセスは利用できません。

次の Transform オブジェクトを例に説明します。

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(ctx, hair_eye_color, processed):
    # データ変換コード
    pass

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

@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

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

@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

データ変換コード内で、以下のようにデータセットのファイルを読み取ることができます。

Copied!1
2
3
4
with hair_eye_color.filesystem().open('students.csv') as f:
   f.readline()

# Result: 'id,hair,eye,sex\n'

ストリームは、パース用のライブラリに渡すこともできます。たとえば、CSV ファイルをパースできます。

Copied!1
2
3
4
5
6
import csv
with hair_eye_color.filesystem().open('students.csv') as f:
    reader = csv.reader(f, delimiter=',')
    next(reader)

# 結果: ['id', 'hair', 'eye', 'sex']

前述のとおり、XML や JSON などの非テーブル形式や、gz や zip などの圧縮形式のファイルも処理できます。たとえば、以下のコードを使用すると、zip ファイル内の CSV ファイルを読み取り、その内容を 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
from transforms.api import Input, Output, transform

import polars as pl
import zipfile


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def read_zip(hair_eye_color, processed):
    fs = hair_eye_color.filesystem()

    def process_zip_file(file_path):
        with fs.open(file_path, "rb") as f:
            with zipfile.ZipFile(f) as archive:
                for filename in archive.namelist():
                    # ディレクトリエントリをスキップします
                    if filename.endswith("/"):
                        continue
                    with archive.open(filename) as f2:
                        df = pl.read_csv(f2, has_header=True)
                        yield df

    all_dfs = []
    for file_status in fs.ls():
        for df in process_zip_file(file_status.path):
            all_dfs.append(df)
    # すべての DataFrame を連結します
    final_df = pl.concat(all_dfs, how="vertical")
    processed.write_table(final_df)

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
from transforms.api import Input, Output, transform

import zipfile
import tempfile


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def read_zip(ctx, hair_eye_color, processed):
    fs = hair_eye_color.filesystem()
    conn = ctx.duckdb().conn

    def process_zip_file(file_path):
        with fs.open(file_path, "rb") as f:
            with zipfile.ZipFile(f) as archive:
                for filename in archive.namelist():
                    # ディレクトリエントリをスキップします
                    if filename.endswith("/"):
                        continue
                    with archive.open(filename) as f2:
                        # DuckDB で読み込めるように一時ファイルに書き込みます
                        with tempfile.NamedTemporaryFile(mode='wb', delete=False, suffix='.csv') as tmp:
                            tmp.write(f2.read())
                            tmp.flush()
                            df = conn.sql(f"SELECT * FROM read_csv('{tmp.name}', header=true)")
                            yield df

    all_queries = []
    for file_status in fs.ls():
        for df in process_zip_file(file_status.path):
            all_queries.append(df)

    # すべてのクエリの和集合を作成します
    if all_queries:
        union_query = ' UNION ALL '.join([f'({query.query})' for query in all_queries])
        final_result = conn.sql(union_query)
        processed.write_table(final_result)

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
from transforms.api import Input, Output, transform

import zipfile
import pandas as pd


@transform.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def read_zip(hair_eye_color, processed):
    fs = hair_eye_color.filesystem()

    def process_zip_file(file_path):
        with fs.open(file_path, "rb") as f:
            with zipfile.ZipFile(f) as archive:
                for filename in archive.namelist():
                    # ディレクトリエントリをスキップします
                    if filename.endswith("/"):
                        continue
                    with archive.open(filename) as f2:
                        df = pd.read_csv(f2, header=0)
                        yield df

    all_dfs = []
    for file_status in fs.ls():
        for df in process_zip_file(file_status.path):
            all_dfs.append(df)

    # すべての DataFrame を連結します
    final_df = pd.concat(all_dfs, axis=0, ignore_index=True)
    processed.write_table(final_df)

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
from transforms.api import transform, Input, Output

import tempfile
import shutil
import zipfile
import io

@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def read_zip(hair_eye_color, processed):
    fs = hair_eye_color.filesystem()
    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()
                with zipfile.ZipFile(tmp) as archive:
                    for filename in archive.namelist():
                        # ディレクトリエントリをスキップします
                        if filename.endswith("/"):
                            continue
                        with archive.open(filename) as f2:
                            br = io.BufferedReader(f2)
                            tw = io.TextIOWrapper(br)
                            tw.readline() # 各 CSV の最初の行をスキップします
                            for line in tw:
                                yield Row(*line.split(","))
    rdd = fs.files().rdd
    rdd = rdd.flatMap(process_file)
    df = rdd.toDF()
    processed.write_dataframe(df)

Excel ファイルをパースする

次の例では、スキーマのないデータセットから 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
from transforms.api import transform, Input, Output
import tempfile
import shutil
import polars as pl
import pandas as pd


@transform.using(
    my_output=Output("/path/tabular_output_dataset"),
    my_input=Input("/path/input_dataset_without_schema"),
)
def compute(my_input, my_output):
    def read_excel_to_polars(fs, file_path):
        with fs.open(file_path, "rb") as f:
            with tempfile.TemporaryFile() as tmp:
                shutil.copyfileobj(f, tmp)
                tmp.flush()  # shutil.copyfileobj はフラッシュしません

                # pd.read_excel() にはシーク可能なファイルが必要です。
                # 列のデータ型が異なると推論されるファイルを
                # 以下で連結できるように、すべての列を文字列として読み込みます。
                pandas_df = pd.read_excel(tmp, dtype=str)
                return pl.from_pandas(pandas_df)

    fs = my_input.filesystem()
    files = [f.path for f in fs.ls()]

    polars_dataframes = []
    for curr_file_path in files:
        polars_dataframes.append(read_excel_to_polars(fs, curr_file_path))

    combined_df = pl.concat(polars_dataframes)
    my_output.write_table(combined_df)

pd.read_excel() にはシークに対応するファイルが必要なため、この例では最初にファイルを一時ファイルにコピーします。通常、.xlsx ファイルの読み取りには openpyxl ライブラリが必要です。環境の依存関係に追加してください。

ランダムアクセス

ランダムアクセスを使用すると、パフォーマンスが大幅に低下します。seek メソッドに依存しないようにコードを書き直すことを推奨します。それでもランダムアクセスを使用したい場合は、以下の方法を参照してください。

open() メソッドはストリームオブジェクトを返すため、ランダムアクセスには対応していません。ランダムアクセスが必要な場合は、ファイルをメモリーまたはディスクにバッファリングできます。hair_eye_color が TransformInput オブジェクトに対応する場合の例を以下に示します。

Copied!1
2
3
4
5
6
7
8
import io
import shutil
s = io.StringIO()
with hair_eye_color.filesystem().open('students.csv') as f:
    shutil.copyfileobj(f, s)
s.getvalue()

# Result: 'id,hair,eye,sex\n...'

Copied!1
2
3
4
5
with hair_eye_color.filesystem().open('students.csv') as f:
    lines = f.read().splitlines()
lines[0]

# 結果: 'id,hair,eye,sex'

Copied!1
2
3
4
5
6
7
8
9
import tempfile
with tempfile.NamedTemporaryFile() as tmp:
    with hair_eye_color.filesystem().open('students.csv', 'rb') as f:
        shutil.copyfileobj(f, tmp)
        tmp.flush()  # shutil.copyfileobj does not flush
    with open(tmp.name) as t:
        t.readline()

# Result: 'id,hair,eye,sex\n'

ファイルに書き込む

ファイルへの書き込みも、同様に open() メソッドを使用して行います。このメソッドは、書き込みのみが可能な Python のファイルライクなストリームオブジェクトを返します。io.open() ↗ が受け付けるすべてのキーワード引数にも対応しています。ファイルはストリームとして書き込まれるため、ランダムアクセスは利用できません。次の Transform オブジェクトを例に説明します。

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

@transform.spark.using(
    hair_eye_color=Input('/examples/students_hair_eye_color'),
    processed=Output('/examples/hair_eye_color_processed')
)
def filter_eye_color(hair_eye_color, processed):
    # type: (TransformInput, TransformOutput) -> None
    # データ変換コード
    pass

データ変換コード内で、出力ファイルシステムに書き込むことができます。たとえば、以下の例のように 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
from transforms.api import transform, Input, Output
import polars as pl


@transform.using(
    processed=Output("/examples/csv_files"),
)
def compute(processed):
    with processed.filesystem().open("csv1.csv", "wb") as f1:
        df1 = pl.DataFrame(
            {
                "student_id": ["001", "002", "003"],
                "hair_color": ["brown", "blonde", "black"],
                "eye_color": ["blue", "green", "brown"],
            }
        )
        df1.write_csv(f1)
    with processed.filesystem().open("csv2.csv", "wb") as f2:
        df2 = pl.DataFrame(
            {
                "student_id": ["004", "005", "006"],
                "hair_color": ["red", "green", "blue"],
                "eye_color": ["purple", "grey", "black"],
            }
        )
        df2.write_csv(f2)

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
import tempfile


@transform.using(
    processed=Output("/examples/csv_files"),
)
def compute(ctx, processed):
    conn = ctx.duckdb().conn

    # 1つ目の CSV ファイルを作成します
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp1:
        conn.execute(f"""COPY (
            SELECT * FROM VALUES
            ('001', 'brown', 'blue'),
            ('002', 'blonde', 'green'),
            ('003', 'black', 'brown')
            AS t(student_id, hair_color, eye_color)
        ) TO '{tmp1.name}' (HEADER)""")

        with open(tmp1.name, 'r') as src:
            with processed.filesystem().open("csv1.csv", "w") as f1:
                f1.write(src.read())

    # 2つ目の CSV ファイルを作成します
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as tmp2:
        conn.execute(f"""COPY (
            SELECT * FROM VALUES
            ('004', 'red', 'purple'),
            ('005', 'green', 'grey'),
            ('006', 'blue', 'black')
            AS t(student_id, hair_color, eye_color)
        ) TO '{tmp2.name}' (HEADER)""")

        with open(tmp2.name, 'r') as src:
            with processed.filesystem().open("csv2.csv", "w") as f2:
                f2.write(src.read())

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
from transforms.api import transform, Input, Output
import pandas as pd


@transform.using(
    processed=Output("/examples/csv_files"),
)
def compute(processed):
    with processed.filesystem().open("csv1.csv", "w") as f1:
        df1 = pd.DataFrame(
            {
                "student_id": ["001", "002", "003"],
                "hair_color": ["brown", "blonde", "black"],
                "eye_color": ["blue", "green", "brown"],
            }
        )
        df1.to_csv(f1)
    with processed.filesystem().open("csv2.csv", "w") as f2:
        df2 = pd.DataFrame(
            {
                "student_id": ["004", "005", "006"],
                "hair_color": ["red", "green", "blue"],
                "eye_color": ["purple", "grey", "black"],
            }
        )
        df2.to_csv(f2)

次の例では、Python 組み込みのシリアライザーである pickle モジュールを使用して、モデルを永続化できます。

Copied!1
2
3
4
import pickle

with processed.filesystem().open('model.pickle', 'wb') as f:
    pickle.dump(model, f)

並列処理

複数のプロセスで処理を行うよう手動で実装することで、ファイル処理を並列化できます。

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
from transforms.api import transform, Input, Output
import polars as pl
from concurrent.futures import ThreadPoolExecutor

@transform.using(
    processed=Output("/Mint/Transforms/lightweight/abort/out"),
    hair_eye_color=Input("/Mint/Transforms/lightweight/abort/csvs"),
)
def example_computation(hair_eye_color, processed):
    fs = hair_eye_color.filesystem()
    # ステップ1：すべての CSV ファイルを一覧表示する
    csv_files = list(fs.ls(glob="*.csv"))

    # ステップ2：1個のファイルを処理する関数
    def process_file(dataset_file):
        path = dataset_file.path
        with fs.open(path) as csv_file:
            # CSV を Polars DataFrame として読み込みます
            df = pl.read_csv(csv_file)
            # 必要に応じて、スキーマに合わせて列を選択するか名前を変更します
            return df.select(["student_id", "hair_color", "eye_color"])

    # ステップ3：ファイルを並列で処理する
    with ThreadPoolExecutor() as executor:
        dfs = list(executor.map(process_file, csv_files))

    # ステップ4：すべての DataFrame を連結する
    result_df = pl.concat(dfs)
    processed.write_table(result_df)

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
from transforms.api import transform, Input, Output
import io
from concurrent.futures import ThreadPoolExecutor

@transform.using(
    processed=Output('/examples/hair_eye_color_processed'),
    hair_eye_color=Input('/examples/students_hair_eye_color_csv'),
)
def example_computation(ctx, hair_eye_color, processed):
    fs = hair_eye_color.filesystem()
    conn = ctx.duckdb().conn

    # ステップ1：すべての CSV ファイルを一覧表示する
    csv_files = list(fs.ls(glob="*.csv"))

    # ステップ2：1個のファイルを処理する関数
    def process_file(dataset_file):
        path = dataset_file.path
        with fs.open(path) as csv_file:
            # CSV の内容をメモリーに読み込みます
            csv_content = csv_file.read()
            df = conn.sql("SELECT student_id, hair_color, eye_color FROM read_csv_auto($1)", [csv_content])
            return df.fetchdf()

    # ステップ3：ファイルを並列で処理する
    with ThreadPoolExecutor() as executor:
        dfs = list(executor.map(process_file, csv_files))

    # ステップ4：DuckDB を使ってすべての DataFrame を連結する
    if dfs:
        # 最初の DataFrame から一時テーブルを作成します
        conn.register('temp_result', dfs[0])
        result_query = conn.sql("SELECT * FROM temp_result")

        # 残りの DataFrame との和集合を作成します
        for i, df in enumerate(dfs[1:], 1):
            conn.register(f'temp_{i}', df)
            result_query = conn.sql(f"SELECT * FROM ({result_query.query}) UNION ALL SELECT * FROM temp_{i}")

        processed.write_table(result_query)

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
from transforms.api import transform, Input, Output
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

@transform.using(
    processed=Output('/examples/hair_eye_color_processed'),
    hair_eye_color=Input('/examples/students_hair_eye_color_csv'),
)
def example_computation(hair_eye_color, processed):
    fs = hair_eye_color.filesystem()

    # ステップ1：すべての CSV ファイルを一覧表示する
    csv_files = list(fs.ls(glob="*.csv"))

    # ステップ2：1個のファイルを処理する関数
    def process_file(dataset_file):
        path = dataset_file.path
        with fs.open(path) as csv_file:
            # CSV を pandas DataFrame として読み込みます
            df = pd.read_csv(csv_file)
            # 必要に応じて、スキーマに合わせて列を選択するか名前を変更します
            return df[["student_id", "hair_color", "eye_color"]]

    # ステップ3：ファイルを並列で処理する
    with ThreadPoolExecutor() as executor:
        dfs = list(executor.map(process_file, csv_files))

    # ステップ4：すべての DataFrame を連結して書き出す
    result_df = pd.concat(dfs, ignore_index=True)
    processed.write_table(result_df)

PySpark による並列処理

DataFrame ↗ オブジェクトで表現されるデータ変換や、標準の非 Spark トランスフォームで定義されるデータ変換とは異なり、PySpark のファイルベースの変換では、ドライバーのコードとエグゼキューターのコードの違いを理解することが重要です。compute 関数は、単一のマシンであるドライバー上で実行されます。Spark は、DataFrame ↗ の関数を、適切と判断した方法でエグゼキューター（多数のマシン）に自動的に分散します。

files API で分散処理のメリットを得るには、Spark を使って計算を分散させる必要があります。そのために、FileStatus の DataFrame ↗ を作成し、エグゼキューター全体に分散させます。すると、エグゼキューター上の各タスクは、割り当てられたファイルを開いて処理でき、その結果は Spark によって集計されます。

files API は、ls() 関数と同じ引数を受け取る files() 関数を公開しています。ただし、この関数は FileStatus オブジェクトの DataFrame ↗ を返します。この DataFrame はファイルサイズでパーティション分割されるため、ファイルサイズが異なる場合でも計算負荷を均等に分散しやすくなります。パーティション分割は、2つの Spark 設定オプションで制御できます。

spark.sql.files.maxPartitionBytes ↗ は、ファイルの読み取り時に1つのパーティションに格納する最大バイト数です。
spark.sql.files.openCostInBytes ↗ は、ファイルを開く際の推定コストで、同じ時間内にスキャンできるバイト数で表されます。この値をファイルサイズに加算して、パーティション内でそのファイルが使用する合計バイト数を計算します。

これらのプロパティの値を変更するには、カスタムの Transforms プロファイルを作成し、configure() デコレーターを使用してトランスフォームに適用する必要があります。詳細については、Code Repositories のドキュメントにある Transforms プロファイルの定義に関するセクションを参照してください。ここで、例を考えてみます。パースして連結したい CSV ファイルがあるとします。flatMap() ↗ を使い、各 FileStatus オブジェクトに処理関数を適用します。この処理関数は、pyspark.sql.SparkSession.createDataFrame() ↗ に従って行を生成する必要があります。

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
import csv
from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, StringType
from transforms.api import transform, Input, Output

@transform.spark.using(
    processed=Output('/examples/hair_eye_color_processed'),
    hair_eye_color=Input('/examples/students_hair_eye_color_csv'),
)
def example_computation(hair_eye_color, processed):
    def process_file(file_status):
        with hair_eye_color.filesystem().open(file_status.path) as f:
            r = csv.reader(f)
            # ヘッダー行から pyspark.Row を作成します
            header = next(r)
            MyRow = Row(*header)
            for row in r:
                yield MyRow(*row)

    schema = StructType([
        StructField('student_id', StringType(), True),
        StructField('hair_color', StringType(), True),
        StructField('eye_color', StringType(), True),
    ])

    files_df = hair_eye_color.filesystem().files('**/*.csv')
    processed_df = files_df.rdd.flatMap(process_file).toDF(schema)
    processed.write_dataframe(processed_df)

処理結果が0行の場合、スキーマ推論に失敗します

スキーマを渡さずに toDF() を呼び出すことはできますが、ファイル処理の結果が0行の場合、Spark のスキーマ推論に失敗し、ValueError: RDD is empty 例外がスローされます。そのため、常にスキーマを手動で指定することを推奨します。

←

PREVIOUSメディアセットトランスフォーム API リファレンス

NEXT高度な使用方法 / 出力行メタデータ

→