COPY INTO を使用してデータを読み込む

[アーティクル]
11/07/2024

COPY INTO SQL コマンドを使用すると、ファイルの場所から Delta テーブルにデータを読み込むことができます。これは再試行可能なべき等操作であり、ソースの場所にある、既に読み込み済みのファイルはスキップされます。

COPY INTO には次の機能があります。

S3、ADLS Gen2、ABFS、GCS、Unity Catalog ボリュームなど、クラウドストレージから簡単に構成できるファイルまたはディレクトリフィルター。
CSV、JSON、XML、Avro、 ORC、Parquet、テキスト、バイナリファイルなど、複数のソースファイル形式のサポート
既定で 1 回だけ (べき等) のファイル処理
ターゲットテーブルスキーマの推論、マッピング、マージ、および展開

Note

よりスケーラブルで堅牢なファイルインジェストエクスペリエンスを実現するために、Databricks においては SQL ユーザーによるストリーミングテーブルの活用をお勧めします。「Databricks SQL でストリーミングテーブルを使用してデータを読み込む」を参照してください。

警告

COPY INTO は、削除ベクトルのワークスペース設定を考慮します。有効にした場合、Databricks Runtime 14.0 以降を実行しているコンピューティングまたは SQL ウェアハウスで COPY INTO が実行されている場合、ターゲットテーブルで削除ベクトルが有効になります。有効にすると、削除ベクトルは Databricks Runtime 11.3 LTS 以下のテーブルに対するクエリをブロックします。「削除ベクトルとは」と「削除ベクトルの自動有効化」を参照してください。

要件

アカウント管理者は、ユーザーが COPY INTO を使用してデータを読み込む前に、「インジェスト用データアクセスを構成する」の手順に従って、クラウドオブジェクトストレージ内のデータへのアクセスを構成する必要があります。

例: スキーマレス Delta Lake テーブルにデータを読み込む

Note

この機能は、Databricks Runtime 11.3 LTS 以降で使用できます。

COPY_OPTIONS で mergeSchema を true に設定することで、後で COPY INTO コマンド中にスキーマが推論されるように、空のプレースホルダー Delta テーブルを作成できます。

CREATE TABLE IF NOT EXISTS my_table
[COMMENT <table-description>]
[TBLPROPERTIES (<table-properties>)];

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

上記の SQL ステートメントはべき等であり、Delta テーブルに 1 回だけデータを取り込むための実行をスケジュールできます。

注意

空の Delta テーブルは、COPY INTO の外部では使用できません。 INSERT INTO および MERGE INTO では、スキーマレス Delta テーブルへのデータの書き込みがサポートされていません。データが COPY INTO でテーブルに挿入されると、テーブルはクエリを実行することが可能になります。

「COPY INTO のターゲットテーブルの作成」を参照してください。

例: スキーマを設定し、Delta Lake テーブルにデータを読み込む

次の例では、Delta テーブルを作成し、COPY INTO SQL コマンドを使用して Databricks データセットのサンプルデータをテーブルに読み込む方法を示しています。 Azure Databricks クラスターにアタッチされているノートブックから、Python、R、Scala、または SQL のサンプルコードを実行できます。また、Databricks SQL の SQL ウェアハウスに関連付けられているクエリから SQL コードを実行することもできます。

SQL

DROP TABLE IF EXISTS default.loan_risks_upload;

CREATE TABLE default.loan_risks_upload (
  loan_id BIGINT,
  funded_amnt INT,
  paid_amnt DOUBLE,
  addr_state STRING
);

COPY INTO default.loan_risks_upload
FROM '/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet'
FILEFORMAT = PARQUET;

SELECT * FROM default.loan_risks_upload;

-- Result:
-- +---------+-------------+-----------+------------+
-- | loan_id | funded_amnt | paid_amnt | addr_state |
-- +=========+=============+===========+============+
-- | 0       | 1000        | 182.22    | CA         |
-- +---------+-------------+-----------+------------+
-- | 1       | 1000        | 361.19    | WA         |
-- +---------+-------------+-----------+------------+
-- | 2       | 1000        | 176.26    | TX         |
-- +---------+-------------+-----------+------------+
-- ...

Python

table_name = 'default.loan_risks_upload'
source_data = '/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet'
source_format = 'PARQUET'

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" \
  "loan_id BIGINT, " + \
  "funded_amnt INT, " + \
  "paid_amnt DOUBLE, " + \
  "addr_state STRING)"
)

spark.sql("COPY INTO " + table_name + \
  " FROM '" + source_data + "'" + \
  " FILEFORMAT = " + source_format
)

loan_risks_upload_data = spark.sql("SELECT * FROM " + table_name)

display(loan_risks_upload_data)

'''
Result:
+---------+-------------+-----------+------------+
| loan_id | funded_amnt | paid_amnt | addr_state |
+=========+=============+===========+============+
| 0       | 1000        | 182.22    | CA         |
+---------+-------------+-----------+------------+
| 1       | 1000        | 361.19    | WA         |
+---------+-------------+-----------+------------+
| 2       | 1000        | 176.26    | TX         |
+---------+-------------+-----------+------------+
...
'''

R

library(SparkR)
sparkR.session()

table_name = "default.loan_risks_upload"
source_data = "/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet"
source_format = "PARQUET"

sql(paste("DROP TABLE IF EXISTS ", table_name, sep = ""))

sql(paste("CREATE TABLE ", table_name, " (",
  "loan_id BIGINT, ",
  "funded_amnt INT, ",
  "paid_amnt DOUBLE, ",
  "addr_state STRING)",
  sep = ""
))

sql(paste("COPY INTO ", table_name,
  " FROM '", source_data, "'",
  " FILEFORMAT = ", source_format,
  sep = ""
))

loan_risks_upload_data = tableToDF(table_name)

display(loan_risks_upload_data)

# Result:
# +---------+-------------+-----------+------------+
# | loan_id | funded_amnt | paid_amnt | addr_state |
# +=========+=============+===========+============+
# | 0       | 1000        | 182.22    | CA         |
# +---------+-------------+-----------+------------+
# | 1       | 1000        | 361.19    | WA         |
# +---------+-------------+-----------+------------+
# | 2       | 1000        | 176.26    | TX         |
# +---------+-------------+-----------+------------+
# ...

Scala

val table_name = "default.loan_risks_upload"
val source_data = "/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet"
val source_format = "PARQUET"

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" +
  "loan_id BIGINT, " +
  "funded_amnt INT, " +
  "paid_amnt DOUBLE, " +
  "addr_state STRING)"
)

spark.sql("COPY INTO " + table_name +
  " FROM '" + source_data + "'" +
  " FILEFORMAT = " + source_format
)

val loan_risks_upload_data = spark.table(table_name)

display(loan_risks_upload_data)

/*
Result:
+---------+-------------+-----------+------------+
| loan_id | funded_amnt | paid_amnt | addr_state |
+=========+=============+===========+============+
| 0       | 1000        | 182.22    | CA         |
+---------+-------------+-----------+------------+
| 1       | 1000        | 361.19    | WA         |
+---------+-------------+-----------+------------+
| 2       | 1000        | 176.26    | TX         |
+---------+-------------+-----------+------------+
...
*/

クリーンアップするには、次のコードを実行します。これで、テーブルが削除されます。

Python

spark.sql("DROP TABLE " + table_name)

R

sql(paste("DROP TABLE ", table_name, sep = ""))

Scala

spark.sql("DROP TABLE " + table_name)

SQL

DROP TABLE default.loan_risks_upload

メタデータファイルをクリーンアップする

Databricks Runtime 15.2 以降で VACUUM を実行すると、COPY INTO によって作成された参照されていないメタデータファイルをクリーンアップできます。

リファレンス

Databricks Runtime 7.x 以降: COPY INTO

その他の技術情報

Unity Catalog ボリュームまたは外部の場所で COPY INTO を使用してデータを読み込む
サービスプリンシパルで COPY INTO を使用してデータを読み込む
同じ Delta テーブルに対する複数の COPY INTO 操作の例など、一般的な使用パターンについては、「COPY INTO を使用した一般的なデータ読み込みパターン」を参照してください。

次の方法で共有

COPY INTO を使用してデータを読み込む

要件

例: スキーマレス Delta Lake テーブルにデータを読み込む

例: スキーマを設定し、Delta Lake テーブルにデータを読み込む

SQL

Python

R

Scala

Python

R

Scala

SQL

メタデータファイルをクリーンアップする

リファレンス

その他の技術情報

フィードバック

その他のリソース

次の方法で共有

COPY INTO を使用してデータを読み込む

要件

例: スキーマレス Delta Lake テーブルにデータを読み込む

例: スキーマを設定し、Delta Lake テーブルにデータを読み込む

SQL

Python

R

Scala

Python

R

Scala

SQL

メタデータ ファイルをクリーンアップする

リファレンス

その他の技術情報

フィードバック

その他のリソース

メタデータファイルをクリーンアップする