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" に変換

時刻/日付

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

数値

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

バイトストリーム

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

text

String

UTF-8 エンコード文字列

Boolean

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

時刻/日付

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

数値

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

バイトストリーム

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

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

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

json

String

UTF-8 エンコード文字列

Boolean

JSON ブール型

時刻/日付

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

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

数値

JSON 数値型

バイトストリーム

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

text

String

UTF-8 エンコード文字列

Boolean

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

時刻/日付

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

数値

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

バイトストリーム

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

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

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

組み込み JSON 形式

String

UTF-8 エンコード文字列

Boolean

JSON ブール型

時刻/日付

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

数値

JSON 数値

バイトストリーム

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

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

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

組み込み JSON 形式

String

UTF-8 エンコード文字列

Boolean

JSON ブール型

時刻/日付

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

数値

JSON 数値

バイトストリーム

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

設定項目

説明

ログオンモジュール

org.apache.kafka.common.security.plain.PlainLoginModule を設定する必要があります。

username

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

password

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

パラメーター

説明

必須

datasource

データソース名。コードエディタはデータソースの追加をサポートしています。このパラメーターの値は、追加されたデータソースの名前と同じである必要があります。

はい

server

ip:port 形式の Kafka ブローカーサーバーのアドレス。

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

はい

topic

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

はい

column

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

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

• データ列

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

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

• 属性列

  • __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 形式の時間文字列です。[スケジューリングパラメーター] と併用できます。詳細については、「サポートされているスケジューリングパラメーターの形式」をご参照ください。

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

endDateTime

データ消費の終了オフセット。これは時間範囲の右境界 (含まない) です。yyyymmddhhmmss 形式の時間文字列です。[スケジューリングパラメーター] と併用できます。詳細については、「サポートされているスケジューリングパラメーターの形式」をご参照ください。

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

beginOffset

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

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

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

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

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

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

endOffset

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

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

skipExceedRecord

Kafka は public ConsumerRecords<K, V> poll(final Duration timeout) を使用してデータを消費します。1 回の poll 呼び出しで、endOffset または endDateTime の範囲外のデータがフェッチされることがあります。skipExceedRecord は、超過したデータを宛先に書き込むかどうかを制御します。データ消費は自動オフセットコミットを使用するため、以下を推奨します：

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

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

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

partition

Kafka Topic には複数のパーティション (partition) があります。デフォルトでは、データ同期タスクは Topic 内のすべてのパーティションをカバーするオフセット範囲からデータを読み取ります。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 から空のデータをプルした場合 (通常は Topic 内のすべてのデータが読み取られたか、ネットワークまたは Kafka クラスターの可用性の問題が原因)、タスクはすぐに停止します。それ以外の場合は、データが再び読み取られるまで再試行します。

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

stopWhenReachEndOffset

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

• このパラメーターが true に設定され、コンシューマーが Kafka から空のデータをプルした場合、Kafka Topic パーティションの最新データが読み取られたかどうかがチェックされます。すべてのパーティションから最新データが読み取られた場合、タスクはすぐに停止します。それ以外の場合は、Kafka Topic からデータをプルし続けます。

• このパラメーターが 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

ip:port 形式の Kafka サーバーのアドレス。

はい

topic

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

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

はい

valueIndex

Kafka ライターで値として使用される列。指定しない場合、デフォルトですべての列が連結されて値を形成します。区切り文字は 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 で設定されたフィールド名の数より少ない場合、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 Topic 内のパーティションの番号を指定します。これは 0 以上の整数でなければなりません。

いいえ

keyIndex

Kafka ライターでキーとして使用される列。

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

いいえ

keyIndexes

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

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

いいえ

fieldDelimiter

writeMode が text に設定され、valueIndex が設定されていない場合、ソースレコードのすべての列は、このパラメーターで指定された列区切り文字を使用して連結され、Kafka レコードの値を形成します。区切り文字として単一の文字または複数の文字を設定できます。Unicode 文字は \u0001 の形式で設定できます。\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：すべてのレプリカへの書き込み成功の確認応答。

いいえ