CSV または JSON ファイルのスキーマを推論する

Foundry のデータセットは、スキーマがあると最も扱いやすくなります。Foundry では、データセット内のスキーマを適用するボタンを選択することで、CSV または JSON ファイルを含むデータセットに手動でスキーマを追加できます。スキーマを適用するボタンは、データの一部に基づいてスキーマを自動的に推論します。スキーマを適用したら、データセットの表示でスキーマを編集を選択して、列のデータ型を変更したり、追加のパースオプションを適用したりできます。これにより、列数が不揃いな行の削除、エンコーディングの変更、ファイルパス、行のバイトオフセット、インポートのタイムスタンプ、行番号といった列の追加ができます。

データセットの初期ファイルに基づいて静的に適用したスキーマは、データが変化すると古くなることがあります。そのため、半構造化データを処理するトランスフォームパイプラインの最初のステップとして、Spark にスキーマを動的に推論させると便利な場合があります。

パイプラインをビルドするたびにスキーマを動的に推論すると、パフォーマンス上のコストがかかります。そのため、この手法は必要な場合（たとえば、スキーマが変化する可能性がある場合）にのみ、限定的に使用してください。

スキーマの動的な推論は、以下の例のように SparkSession.read ↗ でデータセットのファイルのすべて、または一部を読み込む、多くのユースケースの1つにすぎません。スキーマを動的に推論する動作が実際には不要なユースケースでは、inferSchema を False に設定する（この場合、すべての列が文字列になります）か、そのオプションを省略してスキーマを明示的に渡す ↗ことで、この動作を無効にします。スキーマの自動推論を無効にすると、パフォーマンスと一貫性が大幅に向上します。特に、インクリメンタルバッチ間でスキーマの推論結果が異なると問題が発生する可能性のあるインクリメンタル（差分処理）パイプラインで有効です。

以下に、CSV および JSON 入力の例を示します。

Transforms のデフォルトの出力ファイル形式である Parquet では、自動推論されたスキーマに含まれる可能性のある一部の特殊文字を使用できません。そのため、潜在的な問題を防ぐために、以下の例のように sanitize_schema_for_parquet を使用することを推奨します。

CSV

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

@transform(
    output=Output("/Company/sourceA/parsed/data"),
    raw=Input("/Company/sourceA/raw/data_csv"),
)
def read_csv(ctx, raw, output):
    filesystem = raw.filesystem()
    hadoop_path = filesystem.hadoop_path
    files = [f"{hadoop_path}/{f.path}" for f in filesystem.ls()]
    df = (
        ctx
        .spark_session
        .read
        .option("encoding", "UTF-8")  # UTF-8 がデフォルトです
        .option("header", True)
        .option("inferSchema", True)
        .csv(files)
    )
    output.write_dataframe(sanitize_schema_for_parquet(df))

JSON

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

@transform(
    output=Output("/Company/sourceA/parsed/data"),
    raw=Input("/Company/sourceA/raw/data_json"),
)
def read_json(ctx, raw, output):
    filesystem = raw.filesystem()
    hadoop_path = filesystem.hadoop_path
    files = [f"{hadoop_path}/{f.path}" for f in filesystem.ls()]
    df = (
        ctx
        .spark_session
        .read
        .option("multiline", False)  # False がデフォルトです。各ファイルに改行区切りの JSON オブジェクトではなく単一の JSON オブジェクトが含まれる場合は、True を使用します
        .json(files)
    )
    output.write_dataframe(sanitize_schema_for_parquet(df))

←

PREVIOUS概要

NEXTパイプラインのセキュリティ / 概要

→