DataWorks のデータ統合では、Redis Writer を使用して Redis にデータを書き込むことができます。このトピックでは、DataWorks を使用して Redis データソースにデータをバッチ書き込みする方法について説明します。
制限事項
データインポートタスクは、サーバーレスリソースグループ (推奨) およびデータ統合専用リソースグループで実行できます。
Redis Writer を使用してデータを書き込む際、値が List 型の場合、同期タスクを再実行してもべき等性は保証されません。そのため、値が List 型の場合は、タスクを再実行する前に Redis から対応するデータを手動で削除する必要があります。
重要現在、Redis はブルームフィルターの構成をサポートしていません。重複データを処理するための回避策として、ワークフロー内の同期ノードの前後に Shell、Python、PyODPS などのノードを追加して、重複排除を実行できます。
サポートしているデータの型
Redis は、String、List、Set、ZSet (ソート済みセット)、Hash など、豊富な値のデータ型をサポートしています。Redis の詳細については、redis.io をご参照ください。
データ同期タスクの開発
同期タスクの設定のエントリポイントと手順については、以下の設定ガイドをご参照ください。
設定手順については、「コードレス UI でのタスク設定」および「コードエディタでのタスク設定」をご参照ください。
コードエディタのパラメーターとコードサンプルの完全なリストについては、「付録：コードサンプルとパラメーター」をご参照ください。
付録：コードサンプルとパラメーター
コードエディタを使用したバッチ同期タスクの設定
コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプト形式の要件に基づいて、スクリプト内で関連パラメーターを設定する必要があります。詳細については、「コードエディタの使用」をご参照ください。以下では、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明します。
Writer のコードサンプル
次のコードは、MySQL データベースからデータを読み取り、Redis に書き込むデータ同期タスクのサンプルです。MySQL Reader と Redis Writer の両方のコードを示しています。
{
"type":"job",
"version":"2.0", // バージョン番号。
"steps":[
{ // 以下は Reader のコードサンプルです。Reader のパラメーターの詳細については、対応する Reader プラグインのドキュメントをご参照ください。
"stepType":"mysql",
"parameter": {
"envType": 0,
"datasource": "xc_mysql_demo2",
"column": [
"id",
"value",
"table"
],
"connection": [
{
"datasource": "xc_mysql_demo2",
"table": []
}
],
"where": "",
"splitPk": "",
"encoding": "UTF-8"
},,
"name":"Reader",
"category":"reader"
},
{// 以下は Writer のコードサンプルです。
"stepType":"redis", // Redis Writer のプラグイン名。このパラメーターを redis に設定します。
"parameter":{ // 次のセクションでは、Redis Writer の主要なパラメーターについて説明します。
"expireTime":{ // Redis の値のキャッシュ有効期限。このパラメーターは、秒単位または unixtime 単位で設定できます。"seconds":"1000"
},
"keyFieldDelimiter":"u0001", // Redis キーのデリミタ。
"dateFormat":"yyyy-MM-dd HH:mm:ss",// Redis にデータを書き込む際に使用される日付フォーマット。
"datasource":"xc_mysql_demo2", // データソース名。追加したデータソースの名前と同じである必要があります。
"envType": 0, // 環境タイプ。開発環境：1、本番環境：0。
"writeMode":{ // 書き込みモード。
"type":"string" // 値の型。
"mode":"set", // 特定の値の型に対する書き込みモード。
"valueFieldDelimiter":"u0001", // 値間のデリミタ。
},
"keyIndexes":[0,1], // ソースから Redis へのマッピングに使用します。キーとして使用するソース列を指定します (最初の列は 0 から始まります)。たとえば、ソースの最初の列と 2 番目の列を組み合わせて Redis キーにする場合は、このパラメーターを [0,1] に設定します。
"batchSize":"1000" // 各バッチのレコード数。
"column": [ // Redis の string 型で set 操作を行う場合：この列が設定されていない場合、値のフォーマットはデリミタで区切られた文字列 (CSV 形式) になります。(例：ID=1, name="田中", age=18, sex="男性" の場合、Redis の値の例は "18::男性" となります)。column が次の形式で設定されている場合、Redis の値は JSON フォーマットで書き込まれます。(例：ID=1, name="田中", age=18, sex="男性" の場合、Redis の値の例は {"id":1,"name":"田中","age":18,"sex":"男性"} となります)
{
"name": "id",
"index": "0"
},
{
"name": "name",
"index": "1"
},
{
"name": "age",
"index": "2"
},
{
"name": "sex",
"index": "3"
}
]
},
"name":"Writer",
"category":"writer"
}
],
"setting":{
"errorLimit":{
"record":"0" // エラー数。
},
"speed":{
"throttle":true,// throttle が false に設定されている場合、mbps パラメーターは有効にならず、速度制限は無効になります。throttle が true に設定されている場合、速度制限は有効になります。
"concurrent":1, // ジョブの同時実行数。
"mbps":"12"// 速度制限のレート。1 mbps = 1 MB/s。
}
},
"order":{
"hops":[
{
"from":"Reader",
"to":"Writer"
}
]
}
}Writer のパラメーター
パラメーター | 説明 | 必須 | デフォルト値 |
expireTime | Redis の値のキャッシュ有効期限 (秒単位)。このパラメーターが指定されていない場合、デフォルト値の expireTime は、次の 2 つの方法のいずれかで設定できます。
| いいえ | 0 (0 はデータが期限切れにならないことを示します) |
keyFieldDelimiter | Redis キーのデリミタ。例：key=key1\u0001id。複数のキーを連結する必要がある場合に必須です。キーが 1 つだけの場合は、このパラメーターを省略できます。 | いいえ | \u0001 |
dateFormat | Redis にデータを書き込む際に使用される日付フォーマット：yyyy-MM-dd HH:mm:ss。 | いいえ | N/A |
datasource | データソース名。追加したデータソースの名前と同じである必要があります。 | はい | N/A |
selectDatabase | 書き込み先の Redis データベース ( | いいえ | デフォルトではデータベース 0 |
writeMode | Redis Writer は、Redis にデータを書き込むために、次の 5 つの値の型をサポートしています。
writeMode の設定は、値の型によって若干異なります。詳細については、後述の writeMode パラメーターの説明をご参照ください。 説明 Redis Writer を設定する際、writeMode はサポートされている 5 つのデータ型のいずれかに設定する必要があり、指定できる型は 1 つだけです。このパラメーターを設定しない場合、writeMode はデフォルト値の | いいえ | string |
keyIndexes | キーとして使用するソース列のインデックスを指定します。列インデックスは 0 から始まります (最初の列のインデックスは 0、2 番目の列のインデックスは 1、というようになります)。
説明 keyIndexes を設定すると、Redis Writer は残りの列を値として使用します。特定の列のみをキーとして、他の特定の列を値として同期させたい場合、すべての列を同期する必要はありません。Reader プラグインで column パラメーターを指定して列をフィルタリングできます。 | はい | N/A |
batchSize | 各バッチのレコード数。このパラメーターは、データ同期システムと Redis 間のネットワーク対話の数を大幅に削減し、全体のスループットを向上させることができます。この値が大きすぎると、データ同期プロセスでメモリ不足 (OOM) エラーが発生する可能性があります。 | いいえ | 1,000 |
timeout | Redis へのデータ書き込みのタイムアウト (ミリ秒単位)。 | いいえ | 30,000 |
redisMode | Redis の実行モード。有効な値：
説明 サーバーレスリソースグループ (推奨) およびデータ統合専用リソースグループでサポートされています。 | いいえ | N/A |
column | Redis にデータを書き込むための列設定。Redis の string 型で set 操作を行う場合：
| いいえ | N/A |
writeMode パラメーターの説明
Redis Writer を設定する際、writeMode はサポートされている 5 つのデータ型のいずれかに設定する必要があり、指定できる型は 1 つだけです。このパラメーターを設定しない場合、writeMode はデフォルト値の string を使用します。
値の型 | type パラメーター (必須) | mode パラメーター (必須) | valueFieldDelimiter パラメーター (任意) | writeMode の設定例 |
String | type を | mode は書き込みモードのパラメーターです。値が文字列の場合：
| valueFieldDelimiter は値間のデリミタです。デフォルト値は
| |
List | type を | mode は書き込みモードのパラメーターです。値がリストの場合、次のオプションがあります。
| | |
Set | type を | mode は書き込みモードのパラメーターです。値がセットの場合：
| | |
ZSet (ソート済みセット) | type を | mode は書き込みモードのパラメーターです。値が ZSet (ソート済みセット) の場合：
| このパラメーターを設定する必要はありません。 | 説明 値の型が zset の場合、ソースデータの各行は対応するフォーマットに従う必要があります。各行には、キーに加えて 1 つのスコアと値のペアのみを含めることができ、スコアは値の前に配置する必要があります。これにより、Redis Writer はどの列がスコアに対応し、どの列が値に対応するかを正しく識別できます。 |
Hash | type を | mode は書き込みモードのパラメーターです。値がハッシュの場合：
| このパラメーターを設定する必要はありません。 | 説明 値の型が hash の場合、ソースデータの各行は対応するフォーマットに従う必要があります。各行には、キーに加えて 1 つの属性と値のペアのみを含めることができ、属性は値の前に配置する必要があります。これにより、Redis Writer はどの列が属性に対応し、どの列が値に対応するかを正しく識別できます。 |