COPY INTO を使用した一般的なデータロードパターン

COPY INTO を使用してファイルソースから Delta Lake にデータを読み込むための一般的なパターンについて説明します。

COPY INTOを使用するための多くのオプションがあります。

すべてのオプションの完全なリファレンスについては、 COPY INTO を参照してください。

COPY INTOのターゲット表を作成する

COPY INTO 既存の Delta テーブルをターゲットにする必要があります。 Databricks Runtime 11.3 LTS以降では、次のテーブルのスキーマの設定は、スキーマの進化をサポートする形式ではオプションです。

CREATE TABLE IF NOT EXISTS my_table
[(col_1 col_1_type, col_2 col_2_type, ...)]
[COMMENT <table-description>]
[TBLPROPERTIES (<table-properties>)];

COPY INTOを使用してスキーマを推論するには、追加のオプションを渡す必要があることに注意してください。

COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = <format>
FORMAT_OPTIONS ('inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

次の例では、 my_pipe_data というスキーマレス Delta テーブルを作成し、ヘッダーを含むパイプ区切りの CSV を読み込みます。

CREATE TABLE IF NOT EXISTS my_pipe_data;

COPY INTO my_pipe_data
  FROM 'gs://my-bucket/pipeData'
  FILEFORMAT = CSV
  FORMAT_OPTIONS ('mergeSchema' = 'true',
                  'delimiter' = '|',
                  'header' = 'true')
  COPY_OPTIONS ('mergeSchema' = 'true');

COPY INTOでJSONデータをロードする

次の例では、Google Cloud Storage(GCS)内の 5 つのファイルから my_json_dataという Delta テーブルに JSON データを読み込みます。このテーブルは、 COPY INTO を実行する前に作成する必要があります。いずれかのファイルから既にデータが読み込まれている場合、そのファイルのデータは再読み込みされません。

COPY INTO my_json_data
  FROM 'gs://my-bucket/jsonData'
  FILEFORMAT = JSON
  FILES = ('f1.json', 'f2.json', 'f3.json', 'f4.json', 'f5.json')

-- The second execution will not copy any data since the first command already loaded the data
COPY INTO my_json_data
  FROM 'gs://my-bucket/jsonData'
  FILEFORMAT = JSON
  FILES = ('f1.json', 'f2.json', 'f3.json', 'f4.json', 'f5.json')

COPY INTOで Avro データをロードする

次の例では、 SELECT ステートメントの一部として追加の SQL 式を使用して、GCS に Avro データをロードします。

COPY INTO my_delta_table
  FROM (SELECT to_date(dt) dt, event as measurement, quantity::double
          FROM 'gs://my-bucket/avroData')
  FILEFORMAT = AVRO

COPY INTOでCSVファイルをロードする

次の例では、 gs://bucket/base/path/folder1 の GCS から gs:/bucket/deltaTables/targetの Delta テーブルに CSV ファイルをロードします。

COPY INTO delta.`gs:/bucket/deltaTables/target`
  FROM (SELECT key, index, textData, 'constant_value'
          FROM 'gs://bucket/base/path')
  FILEFORMAT = CSV
  PATTERN = 'folder1/file_[a-g].csv'
  FORMAT_OPTIONS('header' = 'true')

-- The example below loads CSV files without headers in GCS using COPY INTO.
-- By casting the data and renaming the columns, you can put the data in the schema you want
COPY INTO delta.`gs:/bucket/deltaTables/target`
  FROM (SELECT _c0::bigint key, _c1::int index, _c2 textData
        FROM 'gs://bucket/base/path')
  FILEFORMAT = CSV
  PATTERN = 'folder1/file_[a-g].csv'

データのロード中に破損したファイルを無視する

破損の問題が原因で読み取れないデータを読み取ることができない場合は、 FORMAT_OPTIONSで ignoreCorruptFiles を true に設定することで、それらのファイルをスキップできます。

COPY INTO コマンドの結果は、 num_skipped_corrupt_files 列の破損のためにスキップされたファイルの数を返します。このメトリクスは、Delta テーブルで DESCRIBE HISTORY を実行した後の [numSkippedCorruptFiles の下の operationMetrics 列にも表示されます。

破損したファイルは COPY INTOによって追跡されないため、破損が修正された場合は、後続の実行で再読み込みできます。 COPY INTO を VALIDATE モードで実行すると、どのファイルが破損しているかを確認できます。

COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = <format>
[VALIDATE ALL]
FORMAT_OPTIONS ('ignoreCorruptFiles' = 'true')

注

ignoreCorruptFiles Databricks Runtime 11.3 LTS 以降で利用できます。