ダウンロード用のデータセットを準備する

以下のエクスポート手順は高度なワークフローです。アクションメニューを使用して Foundry インターフェースからデータを直接ダウンロードできず、別の Foundry アプリケーションからデータをエクスポートすることもできない場合にのみ実行してください。

このガイドでは、Code Repositories または Pipeline Builder のトランスフォームを使用して、ダウンロード用の CSV を準備する方法を説明します。

データを準備する

ダウンロード用の CSV を準備する最初のステップは、データを絞り込んでクレンジングすることです。以下の手順を実行することを推奨します。

データサンプルがエクスポート可能であることを確認し、データのエクスポート制御ルールに従ってください。具体的には、エクスポートが組織のデータガバナンスポリシーに準拠していることを確認してください。
必要な目的を達成できる範囲で、データをできるだけ小さく絞り込みます。最適なパフォーマンスを得るには、CSV 形式のデータの非圧縮サイズを、デフォルトの HDFS ブロックサイズ（128 MB）より小さくする必要があります。そのためには、必要な列のみを選択し、行数を最小限に抑えてください。特定の値で絞り込むか、任意の行数（たとえば1000行）のランダムサンプルを取得することで、行数を減らせます。128 MB を超える CSV を作成しようとすると、時間がかかる場合があり、処理を成功させるために Spark エグゼキューターのメモリーを追加する必要が生じる場合もあります。
列のデータ型を string に変更します。CSV 形式にはスキーマがないため（列のデータ型とラベルが強制されないため）、すべての列を文字列にキャストすることを推奨します。これは、タイムスタンプ列で特に重要です。

以下の Python（PySpark）サンプルコードは、ニューヨーク市のタクシー運行データセットに上記の原則の一部を適用する例を示しています。

def prepare_input(my_input_df):
    from pyspark.sql import functions as F

    filter_column = "vendor_id"
    filter_value = "CMT"
    df_filtered = my_input_df.filter(filter_value == F.col(filter_column))

    approx_number_of_rows = 1000
    sample_percent = float(approx_number_of_rows) / df_filtered.count()

    df_sampled = df_filtered.sample(False, sample_percent, seed=0)

    important_columns = ["medallion", "tip_amount"]

    return df_sampled.select([F.col(c).cast(F.StringType()).alias(c) for c in important_columns])

同様のロジックと Spark の概念を使用して、Pipeline Builder や、SQL、Java などのほかの Spark API でも準備処理を実装できます。

1個のパーティションに減らして出力形式を設定する

エクスポート用のデータの準備ができたら、ダウンロードする出力ファイルが1個になるように、データを単一のパーティションに再分割または結合する必要があります。次に、出力形式を CSV、または JSON、ORC、Parquet、Avro など、サポートされているほかの出力形式に設定します。以下の例では、Pipeline Builder、Python、SQL でデータを再分割し、出力形式を設定する方法を示します。

repartition(1) と coalesce(1) はどちらもデータを単一のパーティションに減らします。ただし、同じトランスフォーム内で絞り込み操作を行う場合は、coalesce の使用を避けてください。coalesce は上流のタスクをまとめるため、絞り込み操作が分散処理されなくなる可能性があります。

Pipeline Builder

まず、データを再パーティション化データ加工を使用して、データを1個のパーティションに減らします。

Pipeline Builder でデータを1個のパーティションに再分割する

次に、パイプラインの出力設定で、書き込み形式を CSV（またはサポートされているほかの形式）に設定します。

Pipeline Builder で出力形式を CSV に設定する

Python

Copied!1
2
3
4
5
6
7
from transforms.api import transform, Input, Output
@transform(
     output=Output("/path/to/python_csv"),
     my_input=Input("/path/to/input")
)
def my_compute_function(output, my_input):
     output.write_dataframe(my_input.dataframe().repartition(1), output_format="csv", options={"header": "true"})

SQL

CREATE TABLE `/path/to/sql_csv` USING CSV AS SELECT /*+ REPARTITION(1) */ * FROM `/path/to/input`

CSV 生成の追加オプションについては、Spark の公式ドキュメント ↗ を参照してください（Pipeline Builder では、これらのオプションの一部のみが利用可能です）。

上記の例では、出力データセットにダウンロード用の CSV ファイルがちょうど1個含まれるように、パーティションを1個に減らしています。その結果、データ全体が1個の Spark エグゼキューターのメモリーに収まる必要があるため、事前にデータを絞り込むか、サンプリングすることを推奨します。データの絞り込みやサンプリングができず、データが大きすぎて1個のエグゼキューターのメモリーに収まらない場合は、適切な Spark プロファイルを使用して、エグゼキューターのメモリーを増やせます。または、パーティション数に1以外の値を使用することもできます。repartition に1より大きい値を指定すると、1個のエグゼキューターのメモリーに保持する必要があるデータ量は減りますが、出力データセットの CSV ファイルが1個ではなく複数になり、それぞれを個別にダウンロードする必要があります。

ダウンロードするファイルにアクセスする

データセットがビルドされたら、データセットページの詳細タブに移動します。CSV がダウンロード可能なファイルとして表示されるはずです。

ダウンロード可能な CSV