DataWorks:Kafka データソース

単一テーブルからのオフライン読み取り

parameter.groupId と parameter.kafkaConfig.group.id の両方が設定されている場合、parameter.groupId が kafkaConfig パラメーター内の group.id よりも優先されます。

単一テーブルへのリアルタイム書き込み

書き込み操作では、データの重複排除はサポートされていません。オフセットのリセットまたはフェールオーバーの後にタスクが再起動されると、重複したデータが書き込まれる可能性があります。

データベース全体のリアルタイム書き込み

• リアルタイムデータ同期タスクは、Serverless リソースグループ (推奨) および 旧バージョンのデータ統合専用リソースグループ をサポートしています。

• ソーステーブルにプライマリキーがある場合、プライマリキーの値が Kafka レコードのキーとして使用されます。これにより、同じプライマリキーへの変更が同じ Kafka パーティションに順序通りに書き込まれることが保証されます。

• ソーステーブルにプライマリキーがない場合は、2 つのオプションがあります。プライマリキーなしでテーブルを同期するオプションを選択した場合、Kafka レコードのキーは空になります。テーブルの変更が Kafka に順序通りに書き込まれるようにするには、送信先の Kafka トピックに単一のパーティションのみを設定する必要があります。カスタムプライマリキーを選択した場合、1 つ以上の非プライマリキーフィールドの組み合わせが Kafka レコードのキーとして使用されます。

• Kafka クラスターが例外を返した場合でも、同じプライマリキーの変更が同じ Kafka パーティションに順序通りに書き込まれるようにするには、拡張パラメーターフォームに次の設定を追加します。

  {"max.in.flight.requests.per.connection":1,"buffer.memory": 100554432}

  重要

  この設定は、同期性能を大幅に低下させます。性能と、厳密な順序付けおよび信頼性の必要性とのバランスを取る必要があります。

• リアルタイム同期で Kafka に書き込まれるメッセージの全体的なフォーマット、ハートビートメッセージのフォーマット、およびソースデータの変更に対応するメッセージのフォーマットの詳細については、「付録：メッセージフォーマット」をご参照ください。

データの読み取り

DataWorks が Kafka からデータを読み取る際、データを JSON フォーマットで解析できます。次の表に、各データモジュールの処理方法を示します。

データの書き込み

DataWorks は、JSON またはテキスト形式で Kafka にデータを書き込みます。処理ポリシーは、以下の表に示すように同期タスクの種類によって異なります。

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

• 手順の詳細については、「コードレス UI でオフライン同期タスクを設定する」および「コードエディタでオフライン同期タスクを設定する」をご参照ください。

• コードエディタのパラメーターの完全なリストとスクリプトのデモについては、「付録：スクリプトのデモとパラメーターの説明」をご参照ください。

単一テーブルまたはデータベース全体のリアルタイム同期タスクの設定

手順については、「単一テーブルのリアルタイム同期タスクを設定する」および「データベース全体のリアルタイム同期タスクを設定する」をご参照ください。

SSL

Kafka データソースを設定する際に、[特別な認証方法] を [SSL] または [SASL_SSL] に設定すると、Kafka クラスターの SSL 認証が有効になります。クライアントのトラストストア証明書ファイルをアップロードし、トラストストアのパスフレーズを入力する必要があります。

• Kafka クラスターが Alibaba Cloud Kafka インスタンスの場合、「SSL 証明書アルゴリズムのアップグレード手順」を参照して、正しいトラストストア証明書ファイルをダウンロードしてください。トラストストアのパスフレーズは KafkaOnsClient です。

• Kafka クラスターが EMR インスタンスの場合、「Kafka 接続に SSL 暗号化を使用する」を参照して、正しいトラストストア証明書ファイルをダウンロードし、トラストストアのパスフレーズを取得してください。

• セルフマネージドクラスターの場合、正しいトラストストア証明書をアップロードし、正しいトラストストアのパスフレーズを入力する必要があります。

キーストア証明書ファイル、キーストアのパスフレーズ、および SSL パスフレーズは、Kafka クラスターで双方向 SSL 認証が有効になっている場合にのみ必要です。Kafka クラスターサーバーはこれを使用してクライアントの ID を認証します。双方向 SSL 認証は、Kafka クラスターの server.properties ファイルで ssl.client.auth=required が設定されている場合に有効になります。詳細については、「Kafka 接続に SSL 暗号化を使用する」をご参照ください。

GSSAPI

Kafka データソースを設定する際に [Sasl メカニズム] を GSSAPI に設定した場合、3 つの認証ファイルをアップロードする必要があります：JAAS 設定ファイル、Kerberos 設定ファイル、および Keytab ファイル。また、専用リソースグループの DNS/HOST 設定も構成する必要があります。以下のセクションでは、これらのファイルと必要な DNS および HOST 設定について説明します。

• JAAS 設定ファイル

  JAAS ファイルは KafkaClient で始まり、その後にすべての設定項目を中括弧 {} で囲む必要があります：

  • 中括弧内の最初の行は、使用するログオンコンポーネントクラスを定義します。異なる SASL 認証メカニズムでは、ログオンコンポーネントクラスは固定されています。後続の各設定項目は key=value 形式で記述されます。

  • 最後の設定項目を除き、すべての設定項目はセミコロンで終わってはいけません。

  • 最後の設定項目はセミコロンで終わり、閉じ括弧 } の後にもう 1 つセミコロンを付ける必要があります。

  フォーマット要件が満たされていない場合、JAAS 設定ファイルは解析できません。次のコードは、典型的な JAAS 設定ファイルのフォーマットを示しています。xxx プレースホルダーを実際の情報に置き換えてください。

  KafkaClient {
     com.sun.security.auth.module.Krb5LoginModule required
     useKeyTab=true
     keyTab="xxx"
     storeKey=true
     serviceName="kafka-server"
     principal="kafka-client@EXAMPLE.COM";
  };

  設定項目	説明
  ログオンモジュール	com.sun.security.auth.module.Krb5LoginModule に設定する必要があります。
  useKeyTab	true に設定する必要があります。
  keyTab	任意のパスを指定できます。同期タスクの実行時に、システムはデータソース設定時にアップロードした keytab ファイルをローカルパスに自動的にダウンロードします。その後、システムはこのローカルパスを keytab 設定項目に使用します。
  storeKey	クライアントがキーを保存するかどうかを指定します。true または false に設定できます。データ同期には影響しません。
  serviceName	Kafka サーバーの server.properties 設定ファイル内の sasl.kerberos.service.name 設定項目に対応します。必要に応じてこの項目を設定してください。
  principal	Kafka クライアントが使用する Kerberos プリンシパルです。必要に応じて設定し、アップロードした keytab ファイルにこのプリンシパルのキーが含まれていることを確認してください。

• Kerberos 設定ファイル

  Kerberos 設定ファイルには、[libdefaults] と [realms] の 2 つのモジュールが含まれている必要があります。

  • [libdefaults] モジュールは Kerberos 認証パラメーターを指定します。モジュール内の各設定項目は key=value 形式で記述されます。

  • [realms] モジュールはキー配布センター (KDC) のアドレスを指定します。複数のレルムサブモジュールを含むことができます。各レルムサブモジュールは、レルム名の後に等号 (=) が続きます。

  その後に、中括弧で囲まれた設定項目のセットが続きます。各設定項目も key=value 形式で記述されます。次のコードは、典型的な Kerberos 設定ファイルのフォーマットを示しています。xxx プレースホルダーを実際の情報に置き換えてください。

  [libdefaults]
    default_realm = xxx
  
  [realms]
    xxx = {
      kdc = xxx
    }

  設定項目	説明
  [libdefaults].default_realm	Kafka クラスターノードにアクセスする際に使用されるデフォルトのレルムです。これは通常、JAAS 設定ファイルで指定されたクライアントプリンシパルのレルムと同じです。
  その他の [libdefaults] パラメーター	[libdefaults] モジュールは、ticket_lifetime などの他の Kerberos 認証パラメーターを指定できます。必要に応じて設定してください。
  [realms].realm name	JAAS 設定ファイルで指定されたクライアントプリンシパルのレルムおよび [libdefaults].default_realm と同じである必要があります。JAAS 設定ファイルのクライアントプリンシパルのレルムが [libdefaults].default_realm と異なる場合は、2 つのレルムサブモジュールを含める必要があります。これらのサブモジュールは、それぞれ JAAS 設定ファイルのクライアントプリンシパルのレルムと [libdefaults].default_realm に対応している必要があります。
  [realms].realm name.kdc	KDC のアドレスとポートを ip:port 形式で指定します。例：kdc=10.0.0.1:88。ポートを省略した場合、システムはデフォルトポート 88 を使用します。例：kdc=10.0.0.1。

• Keytab ファイル

  keytab ファイルには、JAAS 設定ファイルで指定されたプリンシパルのキーが含まれている必要があり、KDC によって検証可能である必要があります。たとえば、現在の作業ディレクトリに client.keytab という名前のファイルがある場合、次のコマンドを実行して、keytab ファイルに指定されたプリンシパルのキーが含まれているかどうかを確認できます。

  klist -ket ./client.keytab
  
  Keytab name: FILE:client.keytab
  KVNO Timestamp           Principal
  ---- ------------------- ------------------------------------------------------
     7 2018-07-30T10:19:16 te**@**.com (des-cbc-md5)

• 専用リソースグループの DNS および HOST 設定

  Kafka クラスターが Kerberos 認証を使用する場合、KDC は各ノードのホスト名を使用して各ノードのプリンシパルを登録します。クライアントが Kafka クラスターノードに接続すると、ローカルの DNS および HOST 設定を使用してノードのプリンシパルを導き出し、KDC にノードのアクセス認証情報を要求します。Kerberos 認証が有効になっている Kafka クラスターにアクセスするために専用リソースグループを使用する場合、クラスターノードのアクセス認証情報を KDC から取得できるように、DNS および HOST 設定を正しく構成する必要があります：

  • DNS 設定

    専用リソースグループがアタッチされている VPC 内の Kafka クラスターノードの名前解決に PrivateZone インスタンスを使用する場合、IP アドレス 100.100.2.136 および 100.100.2.138 のカスタムルートを VPC アタッチメントに追加できます。これにより、Kafka クラスターノードの PrivateZone 名前解決設定が専用リソースグループに適用されることが保証されます。DataWorks コンソールの左側のナビゲーションウィンドウで、[リソースグループリスト] をクリックします。ご利用の専用リソースグループの [アクション] 列で、[ネットワーク設定] をクリックします。[VPC アタッチメント] タブで、[アクション] 列の [カスタムルート] をクリックします。表示されるダイアログボックスで、[ルートの追加] をクリックし、[宛先タイプ] を [IDC] に、[接続方法] を [直接 IP] に設定し、直接 IP アドレスを入力してから、[ルートの生成] をクリックします。

  • HOST 設定

    専用リソースグループがアタッチされている VPC 内の Kafka クラスターノードの名前解決に PrivateZone インスタンスを使用しない場合は、各 Kafka クラスターノードの IP アドレスとドメイン名のマッピングをホスト設定に追加する必要があります。DataWorks コンソールの左側のナビゲーションウィンドウで、[リソースグループリスト] をクリックします。ご利用のリソースグループを見つけ、[アクション] 列の [ネットワーク設定] をクリックし、[ホスト設定] タブを選択します。[追加] をクリックしてホストドメインを追加します。ホスト設定は DNS 設定よりも優先されます。

PLAIN

Kafka データソースを設定する際に [Sasl メカニズム] を PLAIN に設定した場合、JAAS ファイルは KafkaClient で始まり、その後にすべての設定項目を中括弧 {} で囲む必要があります。

• 中括弧内の最初の行は、使用するログオンコンポーネントクラスを定義します。異なる SASL 認証メカニズムでは、ログオンコンポーネントクラスは固定されています。後続の各設定項目は key=value 形式で記述されます。

• 最後の設定項目を除き、すべての設定項目はセミコロンで終わってはいけません。

• 最後の設定項目はセミコロンで終わる必要があります。閉じ括弧 "}" の後にもセミコロンを追加する必要があります。

フォーマット要件が満たされていない場合、JAAS 設定ファイルは解析できません。次のコードは、典型的な JAAS 設定ファイルのフォーマットを示しています。xxx プレースホルダーを実際の情報に置き換えてください。

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="xxx"
  password="xxx";
};

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプトフォーマット要件に基づいて、スクリプト内の関連パラメーターを設定する必要があります。詳細については、「スクリプトモードの設定」をご参照ください。以下の情報は、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要があるパラメーターについて説明しています。

Reader スクリプトのデモ

次の JSON 設定は、Kafka からデータを読み取ります。

{
    "type": "job",
    "steps": [
        {
            "stepType": "kafka",
            "parameter": {
                "server": "host:9093",
                "column": [
                    "__key__",
                    "__value__",
                    "__partition__",
                    "__offset__",
                    "__timestamp__",
                    "'123'",
                    "event_id",
                    "tag.desc"
                ],
                "kafkaConfig": {
                    "group.id": "demo_test"
                },
                "topic": "topicName",
                "keyType": "ByteArray",
                "valueType": "ByteArray",
                "beginDateTime": "20190416000000",
                "endDateTime": "20190416000006",
                "skipExceedRecord": "true"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {
                "print": false,
                "fieldDelimiter": ","
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": true,//throttle が false の場合、mbps パラメーターは有効にならず、レートは制限されません。throttle が true の場合、レートは制限されます。
            "concurrent": 1,//並行スレッドの数。
            "mbps":"12"//最大転送レート。1 mbps は 1 MB/s に相当します。
        }
    }
}

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

次の表に、kafkaConfig パラメーターを示します。

Writer スクリプトのデモ

次のコードは、Kafka にデータを書き込むための JSON 設定を示しています。

{
  "type":"job",
  "version":"2.0",//バージョン番号。
  "steps":[
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType":"Kafka",//プラグイン名。
      "parameter":{
          "server": "ip:9092", //Kafka のサーバーアドレス。
          "keyIndex": 0, //キーとして使用する列。キャメルケースの命名規則に従い、k は小文字にする必要があります。
          "valueIndex": 1, //値として使用する列。現在、ソースデータから 1 つの列を選択するか、このパラメーターを空にすることができます。空の場合、すべてのソースデータが使用されます。
          //たとえば、ODPS テーブルの 2 番目、3 番目、4 番目の列を kafkaValue として使用するには、新しい ODPS テーブルを作成し、元の ODPS テーブルのデータを新しいテーブルにクリーンアップして統合し、その後、新しいテーブルを同期に使用します。
          "keyType": "Integer", //Kafka キーの型。
          "valueType": "Short", //Kafka 値の型。
          "topic": "t08", //Kafka トピック。
          "batchSize": 1024 //一度に Kafka に書き込むデータ量 (バイト)。
        },
      "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 スクリプトのパラメーター