Snowflake データソース - DataWorks - Alibaba Cloud ドキュメントセンター

DataWorks の Data Integration は、Snowflake からのデータの読み取りと Snowflake へのデータの書き込みをサポートしています。このトピックでは、Snowflake で利用可能なデータ同期機能について説明します。

制限事項

Snowflake Reader と Writer は、snowflake-jdbc:3.20.0 ドライバーを使用します。ドライバーの機能の詳細については、Snowflake JDBC Driver Release Notes をご参照ください。

サポートされるフィールドタイプ

Snowflake Reader と Snowflake Writer は、数値や文字など、リレーショナルデータベースのほとんどの一般的なデータ型をサポートしています。ただし、一部のデータ型はサポートされていません。ご利用のデータ型が特定のデータベースでサポートされていることを確認する必要があります。

フィールドタイプ	オフライン読み取り (Snowflake Reader)	オフライン書き込み (Snowflake Writer)
NUMBER	サポート	サポート
DECIMAL	サポート	サポート
NUMERIC	サポート	サポート
INT	サポート	サポート
INTEGER	サポート	サポート
BIGINT	サポート	サポート
SMALLINT	サポート	サポート
TINYINT	サポート	サポート
BYTEINT	サポート	サポート
FLOAT	サポート	サポート
FLOAT4	サポート	サポート
FLOAT8	サポート	サポート
DOUBLE	サポート	サポート
DOUBLE PRECISION	サポート	サポート
CHAR	サポート	サポート
VARCHAR	サポート	サポート
STRING	サポート	サポート
TEXT	サポート	サポート
BINARY	サポート	サポート
VARBINARY	サポート	サポート
BOOLEAN	サポート	サポート
DATE	サポート	サポート
TIME	サポート	サポート
TIMESTAMP	サポート	サポート
TIMESTAMP_NTZ	サポート	サポート
TIMESTAMP_LTZ	サポート	サポート
TIMESTAMP_TZ	非サポート	サポート
OBJECT	非サポート	非サポート
ARRAY	非サポート	非サポート
VARIANT	非サポート	非サポート
GEOGRAPHY	非サポート	非サポート
GEOMETRY	非サポート	非サポート
VECTOR	非サポート	非サポート
FILE	非サポート	非サポート

データソースの作成

データ同期タスクを開発する前に、DataWorks で対応するデータソースを作成する必要があります。手順の詳細については、データソース管理をご参照ください。設定パラメーターの詳細な説明については、設定ページのツールチップをご参照ください。

データ同期タスクの開発

データ同期タスクを設定するためのエントリーポイントと一般的なプロセスについては、次のガイドをご参照ください。

単一テーブルのオフライン同期タスクの設定

手順の詳細については、コードレス UI で同期タスクを設定するおよびコードエディタで同期タスクを設定するをご参照ください。
コードエディタのすべてのパラメーターとサンプルスクリプトについては、本トピックの付録：スクリプトのデモとパラメーターの説明セクションをご参照ください。

付録：スクリプトのデモとパラメーターの説明

オフラインタスクスクリプトの設定

コードエディタを使用して単一テーブルのオフラインタスクを設定する場合、必要なフォーマットでスクリプトを記述する必要があります。詳細については、コードエディタで同期タスクを設定するをご参照ください。以下のセクションでは、コードエディタでのデータソースのパラメーターについて説明します。

Reader スクリプトのデモ

{
    "transform": false,
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "snowflake",
            "parameter": {
                "schema": "PUBLIC",
                "envType": 0,
                "datasource": "snowflake_datasource",
                "column": [
                    "ID",
                    "NAME"
                ],
                "where": "",
                "table": "table"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {
                "print": true
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": false,
            "concurrent": 2
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Reader スクリプトのパラメーター

パラメーター	説明	必須	デフォルト値
datasource	Snowflake データソースの名前。	はい	なし
table	データを同期する元のテーブルの名前。	はい	なし
schema	読み取る Snowflake テーブルが存在するスキーマ。	はい	なし
column	ソーステーブルから同期する列。JSON 配列を使用してフィールドを記述します。デフォルトでは、[""] のようにすべての列が使用されます。列のサブセットを選択してエクスポートできます。テーブルスキーマとは異なる順序で列をエクスポートできます。 JSON 形式で設定を提供することで、定数を設定できます。以下に例を示します： `["id","1", "'bazhen.csy'", "null", "to_char(a + 1)", "2.3" , "true"]`。 id は通常の列名です。 1 は整数定数です。 'bazhen.csy'* は文字列定数です。 null はヌルポインタです。 to_char(a + 1) は関数式です。 2.3 は浮動小数点数です。 true はブール値です。 column パラメーターには、同期する列を指定する必要があります。空にすることはできません。	はい	なし
splitFactor	分割係数。このパラメーターは、データを同期のために分割するチャンクの数を指定します。複数の同時実行スレッドを設定した場合、データは同時実行数 × splitFactor 個のチャンクに分割されます。たとえば、同時実行数が 5 で splitFactor が 5 の場合、データは 25 (5 × 5) 個のチャンクに分割され、5 つの同時実行スレッドで処理されます。説明このパラメーターは 1 から 100 までの値に設定してください。値が大きすぎると、メモリ不足 (OOM) エラーが発生する可能性があります。
splitPk	Snowflake Reader がデータを抽出する際に、splitPk パラメーターを指定して、指定されたプライマリキー列に基づいてデータをシャーディングできます。これにより、データ同期システムが同時タスクを実行して効率を向上させることができます。 splitPk をテーブルのプライマリキーに設定します。プライマリキー列の値は通常、均等に分散されます。これにより、シャード内のデータホットスポットを防ぐことができます。現在、splitPk パラメーターはデータシャーディングに整数列のみをサポートしています。文字列、浮動小数点数、日付などの他のデータ型の列を指定すると、Snowflake Reader はエラーを報告します。
where	フィルター条件。Snowflake Reader は、指定された column、table、および where パラメーターを組み合わせて SQL 文を作成し、データを抽出します。たとえば、テスト目的で、where 条件を `limit 10` に設定できます。増分同期の場合、where 条件を `gmt_create>${bizdate}` に設定できます。 where 条件により、効率的な増分データ同期が可能になります。このパラメーターを設定しないか、空のままにすると、テーブルからすべてのデータが同期されます。	いいえ	なし
fetchSize	このパラメーターは、データベースサーバーから各バッチで取得するレコード数を指定します。値を大きくすると、Data Integration とサーバー間のネットワーク対話の回数が減り、データ抽出のパフォーマンスが向上します。説明 fetchSize の値が 2048 を超えると、データ同期プロセスでメモリ不足 (OOM) エラーが発生する可能性があります。	いいえ	1024

Writer スクリプトのデモ

{
    "transform": false,
    "type": "job",
    "version": "2.0",
    "steps": [
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType": "snowflake",
            "parameter": {
                "schema": "PUBLIC",
                "envType": 0,
                "datasource": "snowflake_datasource",
                "column": [
                    "ID",
                    "NAME"
                ],
                "table": "TABLE"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "executeMode": null,
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": []
    }
}

Writer スクリプトのパラメーター

パラメーター	説明	必須	デフォルト値
datasource	Snowflake データソースの名前。	はい	なし
table	データを同期する先のテーブルの名前。	はい	なし
schema	書き込み先の Snowflake テーブルが存在するスキーマ。	はい	なし
column	書き込み先のテーブルの列。列名をコンマ (,) で区切ります。例： `"column":["id","name","age"]`。すべての列に順番に書き込むには、アスタリスク () を使用します。例： `"column":[""]`。	はい	なし
batchSize	1 回のバッチで書き込むレコード数。値を大きくすると、Data Integration と Snowflake 間のネットワーク対話が大幅に減少し、全体のスループットが向上します。ただし、値が大きすぎると、Data Integration プロセスでメモリ不足 (OOM) エラーが発生する可能性があります。	いいえ	1024