Delta Live Tablesへのデータの取り込み

[Ingest data into Delta Live Tables \| Databricks on AWS](https://docs.databricks.com/workflows/delta-live-tables/delta-live-tables-data-sources.html)の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

データセットを作成するために、以下の外部データソースを使用することができます。

- Databricksランタイムが直接サポートしているすべての[データソース](https://qiita.com/taka_yayoi/items/09348912a92a24441c40)。
- Azure Data Lake Storage Gen2 (ADLS Gen2)、AWS S3、Google Cloud Storage (GCS)のようなクラウドストレージの任意のファイル。
- DBFSに格納されている任意のファイル。

    クラウドストレージへのアクセスの設定に関しては、[クラウドストレージの設定](https://qiita.com/taka_yayoi/items/834b56b30849d63b52d0#%E3%82%AF%E3%83%A9%E3%82%A6%E3%83%89%E3%82%B9%E3%83%88%E3%83%AC%E3%83%BC%E3%82%B8%E3%81%AE%E8%A8%AD%E5%AE%9A)をご覧ください。

サポートされるファイルフォーマット、特に連続的に到着するデータを操作する[ストリーミングライブテーブル](https://qiita.com/taka_yayoi/items/62983c6d48abd843727a)においては、パイプラインで[Auto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)を使用することをお勧めします。Auto Loaderはスケーラブルで効率的であり、スキーマ推定をサポートしています。

Pythonデータセットでは、Auto Loaderでサポートされていないファイルフォーマットからバッチ処理でデータを読み込むために、Apache Sparkのビルトインの[ファイルデータソース](#apache-sparkファイルソース)を活用することができます。

SQLデータセットでは、Auto Loaderでサポートされていないファイルフォーマットからバッチ処理でデータを読み込むために、Delta Live Tablesの[ファイルソース](#spark-sqlファイルソース)を活用することができます。

# Auto Loader

以下のサンプルでは、CSVファイルとJSONファイルからデータセットを作成するためにAuto Loaderを活用しています。

```py:Python
@dlt.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/databricks-datasets/retail-org/customers/")
  )

@dlt.table
def sales_orders_raw():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("/databricks-datasets/retail-org/sales_orders/")
  )
```

```sql:SQL
CREATE OR REFRESH STREAMING LIVE TABLE customers
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/customers/", "csv")

CREATE OR REFRESH STREAMING LIVE TABLE sales_orders_raw
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/sales_orders/", "json")
```

Auto Loaderではサポートされている[フォーマットオプション](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6#%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB%E3%83%95%E3%82%A9%E3%83%BC%E3%83%9E%E3%83%83%E3%83%88%E3%81%AE%E3%82%AA%E3%83%97%E3%82%B7%E3%83%A7%E3%83%B3)を使用することができます。`map()`関数を用いることで、`cloud_files()`メソッドにさまざまななオプションを指定することができます。オプションはキーバリューのペアであり、キーとバリューは文字列となります。以下では、SQLでAuto Loaderを操作する際の構文を示しています。

```sql:SQL
CREATE OR REFRESH STREAMING LIVE TABLE <table_name>
AS SELECT *
  FROM cloud_files(
    "<file_path>",
    "<file_format>",
    map(
      "<option_key>", "<option_value",
      "<option_key>", "<option_value",
      ...
    )
  )
```

以下のサンプルでは、ヘッダーありのタブ区切りのCSVファイルからデータを読み込んでいます。

```sql:SQL
CREATE OR REFRESH STREAMING LIVE TABLE customers
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/customers/", "csv", map("delimiter", "\t", "header", "true"))
```

手動でフォーマットを指定するために`schema`を使用することができます。[スキーマ推定](https://qiita.com/taka_yayoi/items/2ccef74b279b1477154b)をサポートしていないフォーマットでは`schema`を指定しなくてはなりません。

```py:Python
@dlt.table
def wiki_raw():
  return (
    spark.readStream.format("cloudFiles")
      .schema("title STRING, id INT, revisionId INT, revisionTimestamp TIMESTAMP, revisionUsername STRING, revisionUsernameId INT, text STRING")
      .option("cloudFiles.format", "parquet")
      .load("/databricks-datasets/wikipedia-datasets/data-001/en_wikipedia/articles-only-parquet")
  )
```

```sql:SQL
CREATE OR REFRESH STREAMING LIVE TABLE wiki_raw
AS SELECT *
  FROM cloud_files(
    "/databricks-datasets/wikipedia-datasets/data-001/en_wikipedia/articles-only-parquet",
    "parquet",
    map("schema", "title STRING, id INT, revisionId INT, revisionTimestamp TIMESTAMP, revisionUsername STRING, revisionUsernameId INT, text STRING")
  )
```

:::note
**注意**
ファイルを読み込むためにAuto Loaderを使用する際、Delta Live Tablesは自動でスキーマやチェックポイントのディレクトリを設定、管理します。これらのディレクトリのいずれかを手動で設定し、フルリフレッシュを実行しても、設定されたディレクトリのコンテンツには影響を及ぼしません。処理過程で期待しない副作用を避けるために、自動で設定されるディレクトリを使用することをお勧めします。
:::

# Apache Sparkファイルソース

Pythonでデータセットを定義し、バッチ処理でファイルを読み込むには、標準的なPySpark関数を使用することができます。以下のサンプルでは、PySparkの`spark.read.format("parquet").load()`関数を用いて、ファイルからParquetデータを読み込んでいます。

```py:Python
@dlt.table
def lendingclub_raw_data():
  return (
    spark.read.format("parquet").load("/databricks-datasets/samples/lending_club/parquet/")
  )
```

# Spark SQLファイルソース

SQLでデータセットを定義し、バッチ処理でファイルを読み込むには、[Spark SQL](https://qiita.com/taka_yayoi/items/47b58eb8696b278a3a82#%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%AE%E8%AA%AD%E3%81%BF%E8%BE%BC%E3%81%BF)の構文を使用することができます。以下のサンプルでは、ファイルからParquetデータを読み込んでいます。

```sql:SQL
CREATE OR REFRESH LIVE TABLE customers
AS SELECT * FROM parquet.`/databricks-datasets/samples/lending_club/parquet/`
```

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)