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

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

Time/Date

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

Numeric

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

Byte stream

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

text

String

UTF-8 エンコード済み文字列

Boolean

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

Time/Date

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

Numeric

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

Byte stream

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

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

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

json

String

UTF-8 エンコード済み文字列

Boolean

JSON ブール型

Time/Date

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

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

Numeric

JSON 数値型

Byte stream

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

text

String

UTF-8 エンコード済み文字列

Boolean

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

Time/Date

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

Numeric

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

Byte stream

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

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

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

組み込み JSON 形式

String

UTF-8 エンコード済み文字列

Boolean

JSON ブール型

Time/Date

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

Numeric

JSON 数値

Byte stream

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

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

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

組み込み JSON 形式

String

UTF-8 エンコード済み文字列

Boolean

JSON ブール型

Time/Date

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

Numeric

JSON 数値

Byte stream

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

設定項目

説明

ログインモジュール

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

username

ユーザー名です。必要に応じて設定してください。

password

パスワードです。必要に応じて設定してください。

パラメーター

説明

必須

datasource

データソースの名称です。コードエディタではデータソースの追加が可能です。このパラメーターの値は、追加したデータソースの名称と一致している必要があります。

はい

server

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

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

はい

topic

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

はい

column

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

• 定数列：シングルクォートで囲まれた列（例：["'abc'", "'123'"]）。

• データ列

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

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

• 属性列

  • __key__：メッセージのキー。

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

  • __partition__：現在のメッセージが存在する partition。

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

  • __offset__：現在のメッセージの offset。

  • __timestamp__：現在のメッセージの 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 の保存済み offset からデータを読み取ります。注：グループオフセットは定期的にコミットされます。タスクが失敗して再実行された場合、データが重複または欠落する可能性があります。skipExceedRecord パラメーターを true に設定した場合、最後の数レコードが破棄される可能性があります。この破棄されたデータのオフセットはすでにコミットされているため、次回の実行では再読み取りできません。

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

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

endOffset

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

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

skipExceedRecord

Kafka は public ConsumerRecords<K, V> poll(final Duration timeout) メソッドを使用してデータを消費します。1 回の poll 呼び出しで取得されるデータは、endOffset または endDateTime を超える場合があります。skipExceedRecord パラメーターは、この超過データを送信先に書き込むかどうかを制御します。オフセットは自動的にコミットされるため、以下のとおり推奨します。

• Kafka のバージョンが 0.10.2 より前の場合：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

コンシューマーオブジェクトが Kafka からデータを 1 回の試行で取得する際の最大待機時間（秒）です。

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

stopWhenPollEmpty

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

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

stopWhenReachEndOffset

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

• true：コンシューマーが Kafka から空のデータを取得した場合、システムは Kafka トピックパーティションの最新 offset がすべて読み取られたかどうかをチェックします。読み取られている場合、タスクは直ちに停止します。そうでない場合、タスクは Kafka トピックから引き続きデータを取得します。

• false：コンシューマーが Kafka から空のデータを取得した場合、システムはチェックを行わず、タスクを直ちに停止します。

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

パラメーター

説明

fetch.min.bytes

コンシューマーがブローカーから取得できるデータの最小量（バイト単位）です。十分な量のデータが利用可能になった場合にのみ、データがコンシューマーに返されます。

fetch.max.wait.ms

ブローカーがデータを返すまでの最大待機時間です。デフォルト値は 500 ms です。ブローカーは、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 ライターで値として使用する列です。指定しない場合、デフォルトですべての列が連結されて値が形成されます。区切り文字は 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 ライターでキーとして使用する列です。

keyIndex パラメーターの値は 0 以上の整数である必要があります。それ以外の場合、タスクは失敗します。

いいえ

keyIndexes

ソースレコードの列の序数（0 から始まる）の配列で、Kafka レコードのキーとして使用します。

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

いいえ

fieldDelimiter

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

いいえ