DataWorks:Kafka データソース

Kafka レコードデータモジュール

処理されるデータの型

key

データ同期タスクの keyType 設定項目に依存します。keyType パラメーターの詳細については、付録の完全なパラメーター説明をご参照ください。

value

データ同期タスクの valueType 設定項目に依存します。valueType パラメーターの詳細については、付録の完全なパラメーター説明をご参照ください。

offset

Long

timestamp

Long

headers

String

partition

Long

同期タスクのタイプ

Kafka に書き込まれる value のフォーマット

ソースフィールドの型

書き込み操作の処理方法

オフライン同期

DataStudio のオフライン同期ノード

json

String

UTF-8 エンコードされた文字列

Boolean

UTF-8 エンコードされた文字列 "true" または "false" に変換

Time/Date

yyyy-MM-dd HH:mm:ss 形式の UTF-8 エンコードされた文字列

Numeric

UTF-8 エンコードされた数値文字列

バイトストリーム

バイトストリームは UTF-8 エンコードされた文字列として扱われ、文字列に変換されます。

text

String

UTF-8 エンコードされた文字列

Boolean

UTF-8 エンコードされた文字列 "true" または "false" に変換

Time/Date

yyyy-MM-dd HH:mm:ss 形式の UTF-8 エンコードされた文字列

Numeric

UTF-8 エンコードされた数値文字列

バイトストリーム

バイトストリームは UTF-8 エンコードされた文字列として扱われ、文字列に変換されます。

リアルタイム同期：Kafka へのリアルタイム ETL

DataStudio のリアルタイム同期ノード

json

String

UTF-8 エンコードされた文字列

Boolean

JSON Boolean 型

Time/Date

• ミリ秒未満の精度の時間値の場合：ミリ秒単位のタイムスタンプを表す 13 桁の JSON 整数に変換されます。

• マイクロ秒またはナノ秒の精度の時間値の場合：ミリ秒のタイムスタンプを表す 13 桁の整数と、ナノ秒のタイムスタンプを表す 6 桁の小数を含む JSON 浮動小数点数に変換されます。

Numeric

JSON 数値型

バイトストリーム

バイトストリームは Base64 エンコードされた後、UTF-8 エンコードされた文字列に変換されます。

text

String

UTF-8 エンコードされた文字列

Boolean

UTF-8 エンコードされた文字列 "true" または "false" に変換

Time/Date

yyyy-MM-dd HH:mm:ss 形式の UTF-8 エンコードされた文字列

Numeric

UTF-8 エンコードされた数値文字列

バイトストリーム

バイトストリームは Base64 エンコードされた後、UTF-8 エンコードされた文字列に変換されます。

リアルタイム同期：データベース全体の Kafka へのリアルタイム同期

増分データのみのリアルタイム同期

組み込み JSON 形式

String

UTF-8 エンコードされた文字列

Boolean

JSON Boolean 型

Time/Date

13 桁のミリ秒タイムスタンプ

Numeric

JSON 数値

バイトストリーム

バイトストリームは Base64 エンコードされた後、UTF-8 エンコードされた文字列に変換されます。

同期ソリューション：Kafka へのワンクリックリアルタイム同期

完全オフライン同期 + 増分リアルタイム同期

組み込み JSON 形式

String

UTF-8 エンコードされた文字列

Boolean

JSON Boolean 型

Time/Date

13 桁のミリ秒タイムスタンプ

Numeric

JSON 数値

バイトストリーム

バイトストリームは Base64 エンコードされた後、UTF-8 エンコードされた文字列に変換されます。

設定項目

説明

ログオンモジュール

org.apache.kafka.common.security.plain.PlainLoginModul に設定する必要があります

username

ユーザー名。必要に応じてこの項目を設定してください。

password

パスワード。必要に応じてこの項目を設定してください。

パラメーター

説明

必須

datasource

データソースの名前。コードエディタはデータソースの追加をサポートしています。このパラメーターの値は、追加されたデータソースの名前と同じでなければなりません。

はい

server

Kafka ブローカーサーバーのアドレスを ip:port 形式で指定します。

server は 1 つしか設定できませんが、DataWorks が Kafka クラスター内のすべてのブローカーの IP アドレスに接続できることを確認する必要があります。

はい

topic

Kafka トピック。トピックは、Kafka が処理するメッセージフィードの集約です。

はい

column

読み取る Kafka データ。定数列、データ列、属性列がサポートされています。

• 定数列：単一引用符で囲まれた列。例：["'abc'", "'123'"]。

• データ列

  • データが JSON 形式の場合、JSON オブジェクトのプロパティを取得できます。例：["event_id"]。

  • データが JSON 形式の場合、JSON オブジェクトのネストされたサブプロパティを取得できます。例：["tag.desc"]。

• 属性列

  • __key__：メッセージキー。

  • __value__：メッセージの完全な内容。

  • __partition__：現在のメッセージが存在するパーティション。

  • __headers__：現在のメッセージのヘッダー。

  • __offset__：メッセージオフセット。

  • __timestamp__：メッセージのタイムスタンプ。

  以下のコードは完全な例です。

  "column": [
      "__key__",
      "__value__",
      "__partition__",
      "__offset__",
      "__timestamp__",
      "'123'",
      "event_id",
      "tag.desc"
      ]

はい

keyType

Kafka キーの型。有効な値：BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG、SHORT。

いいえ

valueType

Kafka 値の型。有効な値：BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG、SHORT。

いいえ

beginDateTime

データ消費の開始時刻。このパラメーターは時間範囲の左境界を指定し、包括的です。値は yyyymmddhhmmss 形式の時間文字列でなければなりません。このパラメーターは scheduling parameters と共に使用できます。詳細については、「サポートされるスケジューリングパラメーターの形式」をご参照ください。

このパラメーターまたは beginOffset のいずれかを指定する必要があります。

endDateTime

データ消費の終了時刻。このパラメーターは時間範囲の右境界を指定し、排他的です。値は yyyymmddhhmmss 形式の時間文字列でなければなりません。このパラメーターは scheduling parameters と共に使用できます。詳細については、「サポートされるスケジューリングパラメーターの形式」をご参照ください。

このパラメーターまたは endOffset のいずれかを指定する必要があります。

beginOffset

データ消費の開始オフセット。以下の形式で設定できます：

• 数値、例：15553274。これは消費の開始オフセットを示します。

• seekToBeginning：最小オフセットからデータを消費することを示します。

• seekToLast：コンシューマーグループの最後にコミットされたオフセットからデータを読み取ります。グループ ID は kafkaConfig 設定の group.id パラメーターで指定されます。クライアントは自動的にオフセットをコミットします。そのため、タスクが失敗して再実行されると、データが重複したり失われたりする可能性があります。skipExceedRecord パラメーターが true に設定されている場合、タスクは最後に読み取ったいくつかのレコードを破棄する可能性がありますが、このデータのオフセットはすでにコミットされているため、次回の実行では読み取られません。

• seekToEnd：最大オフセットからデータを消費することを示します。これにより、空のデータが読み取られます。

このパラメーターまたは beginDateTime のいずれかを指定する必要があります。

endOffset

データ消費の終了オフセット。これは、データ消費タスクがいつ終了するかを制御するために使用されます。

このパラメーターまたは endDateTime のいずれかを指定する必要があります。

skipExceedRecord

Kafka は public ConsumerRecords<K, V> poll(final Duration timeout) を使用してデータを消費します。1 回の poll 呼び出しで、endOffset または endDateTime で指定された範囲外のデータをフェッチする可能性があります。このパラメーターは、この超過データを処理するか破棄するかを制御します。データ消費には自動オフセットコミットが使用されるため、以下のルールに基づいてこのパラメーターを設定することを推奨します：

• 0.10.2 より前の Kafka バージョンの場合：skipExceedRecord を false に設定します。

• Kafka 0.10.2 以降の場合：skipExceedRecord を true に設定します。

いいえ。デフォルト値は false です。

partition

Kafka トピックには複数のパーティション (partition) があります。デフォルトでは、データ同期タスクはトピック内のすべてのパーティションをカバーするオフセット範囲からデータを読み取ります。partition を指定して、単一のパーティションのオフセット範囲からのみデータを読み取ることもできます。

いいえ。デフォルト値はありません。

kafkaConfig

データ消費のために KafkaConsumer クライアントを作成する際、bootstrap.servers、auto.commit.interval.ms、session.timeout.ms などの拡張パラメーターを指定できます。kafkaConfig を使用して、KafkaConsumer の消費動作を制御できます。

いいえ

encoding

keyType または valueType が STRING に設定されている場合、このパラメーターで指定されたエンコーディングが文字列の解析に使用されます。

いいえ。デフォルト値は UTF-8 です。

waitTIme

コンシューマーオブジェクトが 1 回の試行で Kafka からデータをプルするのを待つ最大時間 (秒)。

いいえ。デフォルト値は 60 です。

stopWhenPollEmpty

有効な値は true と false です。このパラメーターが true に設定され、コンシューマーが Kafka から空のデータをプルした場合 (通常はトピック内のすべてのデータが読み取られたため、またはネットワークや Kafka クラスターの可用性の問題のため)、タスクは直ちに停止します。それ以外の場合、データが再び読み取られるまで再試行します。

いいえ。デフォルト値は true です。

stopWhenReachEndOffset

このパラメーターは、stopWhenPollEmpty が true の場合にのみ有効です。有効な値は true と false です。

• このパラメーターが true に設定され、コンシューマーが Kafka から空のデータをプルした場合、システムはトピックパーティションの最大オフセットが読み取られたかどうかを確認します。すべてのパーティションの最大オフセットが読み取られた場合、タスクは直ちに停止します。それ以外の場合、タスクは Kafka トピックからデータのプルを続行します。

• このパラメーターが false に設定され、コンシューマーが Kafka から空のデータをプルした場合、システムはチェックを実行せずにタスクを直ちに停止します。

いいえ。デフォルト値は false です。

パラメーター

説明

fetch.min.bytes

コンシューマーがブローカーからフェッチできる最小データ量 (バイト)。指定された量のデータが利用可能になった場合にのみ、データがコンシューマーに返されます。

fetch.max.wait.ms

サーバーがブローカーからデータが返されるのを待つ最大時間。デフォルト値は 500 ミリ秒です。データは、fetch.min.bytes と fetch.max.wait.ms パラメーターで指定された条件のいずれかが最初に満たされた時点で返されます。

max.partition.fetch.bytes

ブローカーが各 partition からコンシューマーに返すことができる最大バイト数を指定します。デフォルト値は 1 MB です。

session.timeout.ms

コンシューマーがサービスを受けられなくなる前にサーバーから切断できる時間を指定します。デフォルト値は 30 秒です。

auto.offset.reset

オフセットがないか、無効なオフセットで読み取る場合 (コンシューマーが長時間非アクティブで、オフセットを持つレコードが期限切れで削除されたため) にコンシューマーが取るアクション。デフォルト値は none で、オフセットは自動的にリセットされません。earliest に変更でき、これはコンシューマーが最小オフセットから partition レコードを読み取ることを意味します。

max.poll.records

poll メソッドの 1 回の呼び出しで返されるメッセージの数。

key.deserializer

メッセージキーの逆シリアル化メソッド。例：org.apache.kafka.common.serialization.StringDeserializer。

value.deserializer

データ値の逆シリアル化メソッド。例：org.apache.kafka.common.serialization.StringDeserializer。

ssl.truststore.location

SSL ルート証明書のパス。

ssl.truststore.password

ルート証明書ストアのパスワード。Alibaba Cloud Kafka を使用している場合は、これを KafkaOnsClient に設定します。

security.protocol

アクセスプロトコル。現在、SASL_SSL プロトコルのみがサポートされています。

sasl.mechanism

SASL 認証方式。Alibaba Cloud Kafka を使用している場合は、PLAIN を使用します。

java.security.auth.login.config

SASL 認証ファイルのパス。

パラメーター

説明

必須

datasource

データソースの名前。コードエディタはデータソースの追加をサポートしています。このパラメーターの値は、追加されたデータソースの名前と同じでなければなりません。

はい

server

Kafka サーバーのアドレスを ip:port 形式で指定します。

はい

topic

Kafka トピック。Kafka が処理するさまざまなメッセージフィードのカテゴリです。

Kafka クラスターに公開される各メッセージにはカテゴリがあり、これをトピックと呼びます。トピックはメッセージのグループのコレクションです。

はい

valueIndex

Kafka writer で値として使用される列。これを指定しない場合、デフォルトですべての列が連結されて値を形成します。区切り文字は fieldDelimiter で指定されます。

いいえ

writeMode

valueIndex が設定されていない場合、このパラメーターはソースレコードのすべての列を連結して Kafka レコードの値を形成するフォーマットを決定します。有効な値は text と JSON です。デフォルト値は text です。

• text に設定すると、すべての列が fieldDelimiter で指定された区切り文字を使用して連結されます。

• JSON に設定すると、すべての列が column パラメーターで指定されたフィールド名に基づいて JSON 文字列に連結されます。

たとえば、ソースレコードに a、b、c の値を持つ 3 つの列があり、writeMode が text に、fieldDelimiter が # に設定されている場合、書き込まれる Kafka レコードの値は文字列 a#b#c になります。writeMode が JSON に、column が [{"name":"col1"},{"name":"col2"},{"name":"col3"}] に設定されている場合、書き込まれる Kafka レコードの値は文字列 {"col1":"a","col2":"b","col3":"c"} になります。

valueIndex が設定されている場合、このパラメーターは無効です。

いいえ

column

データを書き込む送信先テーブルのフィールドをカンマで区切って指定します。例："column": ["id", "name", "age"]。

valueIndex が設定されておらず、writeMode が JSON に設定されている場合、このパラメーターはソースレコードの列値の JSON 構造内のフィールド名を定義します。例："column": [{"name":id","type":"JSON_NUMBER"}, {"name":"name","type":"JSON_STRING"}, {"name":"age","type":"JSON_NUMBER"}]。

• ソースレコードの列数が column で設定されたフィールド名の数より多い場合、書き込み中にデータは切り捨てられます。例：

  ソースレコードに a、b、c の値を持つ 3 つの列があり、column が [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"}] として設定されている場合、書き込まれる Kafka レコードの値は文字列 {"col1":"a","col2":"b"} になります。

• ソースレコードの列数が column で指定したフィールド名の数より少ない場合、余分なフィールド名は null または nullValueFormat で指定した文字列で埋められます。例：

  ソースレコードに a と b の値を持つ 2 つの列があり、column が [{"name":"col1","type":"JSON_STRING"},{"name":"col2","type":"JSON_STRING"},{"name":"col3","type":"JSON_STRING"}] として設定されている場合、書き込まれる Kafka レコードの値は文字列 {"col1":"a","col2":"b","col3":null} になります。valueIndex が設定されているか、writeMode が text に設定されている場合、このパラメーターは無効です。

• JSON フィールドタイプが設定されていない場合、デフォルトのフィールドタイプは JSON_STRING です。

• JSON フィールドタイプの有効な値については、「付録：JSON フィールドタイプ」をご参照ください。

valueIndex が設定されているか、writeMode が text に設定されている場合、このパラメーターは無効です。

valueIndex が設定されておらず、writeMode が JSON に設定されている場合に必須です。

partition

データが書き込まれる Kafka トピックのパーティション番号を指定します。これは 0 以上の整数でなければなりません。

いいえ

keyIndex

Kafka writer でキーとして使用される列。

keyIndex パラメーターの値は 0 以上の整数でなければなりません。そうでない場合、タスクは失敗します。

いいえ

keyIndexes

Kafka レコードのキーとして使用されるソースレコード内の列の序数の配列。

列の序数は 0 から始まります。たとえば、[0,1,2] は、設定されたすべての列番号の値をカンマで連結して Kafka レコードのキーを形成します。これを指定しない場合、Kafka レコードのキーは null になり、データはトピックのパーティションにラウンドロビン方式で書き込まれます。このパラメーターまたは keyIndex のいずれか一方のみを指定できます。

いいえ

fieldDelimiter

writeMode が text に設定され、valueIndex が設定されていない場合、ソースレコードのすべての列がこのパラメーターで指定された列区切り文字を使用して連結され、Kafka レコードの値を形成します。区切り文字として単一の文字または複数の文字を設定できます。\u0001 形式で Unicode 文字を設定できます。\t や \n などのエスケープ文字がサポートされています。デフォルト値は \t です。

writeMode が text に設定されていないか、valueIndex が設定されている場合、このパラメーターは無効です。

いいえ

keyType

Kafka キーの型。有効な値：BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG、SHORT。

はい

valueType

Kafka 値の型。有効な値：BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG、SHORT。

はい

nullKeyFormat

keyIndex または keyIndexes で指定されたソース列の値が null の場合、このパラメーターで指定された文字列に置き換えられます。これを設定しない場合、置き換えは行われません。

いいえ

nullValueFormat

ソース列の値が null の場合、Kafka レコードの値がアセンブルされるときに、このパラメーターで指定した文字列に値が置き換えられます。このパラメーターを指定しない場合、置き換えは行われません。

いいえ

acks

Kafka プロデューサーを初期化する際の acks 設定。これは、書き込み成功の確認応答方式を決定します。デフォルトでは、acks パラメーターは all に設定されています。acks の有効な値は次のとおりです：

• 0：書き込み成功の確認応答なし。

• 1：プライマリレプリカへの書き込み成功の確認応答。

• all：すべてのレプリカへの書き込み成功の確認応答。

いいえ