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 つのファイルから JSON データを Delta テーブル my_json_data
に読み込みます。 このテーブルは、 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 から Delta テーブルに CSV ファイルを読み込みます。
COPY INTO target_table
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 target_table
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
によって追跡されないため、破損が修正された場合は、後続の実行で再読み込みできます。 VALIDATE
モードでCOPY INTO
を実行すると、どのファイルが破損しているかを確認できます。
COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = <format>
[VALIDATE ALL]
FORMAT_OPTIONS ('ignoreCorruptFiles' = 'true')
注:
ignoreCorruptFiles
は、Databricks Runtime 11.3 LTS 以降で使用できます。