DataWorks：Kafka資料來源

單表離線讀

同時配置parameter.groupId和parameter.kafkaConfig.group.id時，parameter.groupId優先順序高於kafkaConfig配置資訊中的group.id。

單表即時寫

寫入資料不支援去重，即如果任務重設位點或者Failover後再啟動，會導致出現重複資料寫入。

整庫即時寫

• 即時資料同步任務支援使用Serverless資源群組（推薦）和舊版獨享Data Integration資源群組。

• 對於源端同步表有主鍵的情境，同步時會使用主索引值作為Kafka記錄的key，確保同主鍵的變更有序寫入Kafka的同一分區。

• 對於源端同步表無主鍵的情境，如果選擇了支援無主鍵表同步選項，則同步時Kafka記錄的key為空白。如果要確保表的變更有序寫入Kafka，則選擇寫入的Kafka topic必須是單分區。如果選擇了自訂同步主鍵，則同步時使用其他非主鍵的一個或幾個欄位的聯合，代替主鍵作為Kafka記錄的key。

• 如果在Kafka叢集發生響應異常的情況下，仍要確保有主鍵表同主鍵的變更有序寫入Kafka的同一分區，則需要在配置Kafka資料來源時，在擴充參數表單中加入如下配置。

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

  重要

  添加配置後同步效能會大幅下降，需要在效能和嚴格保序可靠性之間做好權衡。

• 即時同步寫入Kafka的訊息總體格式、同步任務心跳訊息格式及源端更改資料對應的訊息格式，詳情請參見：附錄：訊息格式。

讀資料

DataWorks讀取Kafka資料時，支援對Kafka資料進行JSON格式的解析，各資料模組的處理方式如下。

寫資料

DataWorks將資料寫入Kafka時，支援寫入JSON格式或text格式的資料，不同的資料同步任務類型往Kafka資料來源中寫入資料時，對資料的處理策略不一致，詳情如下。

單表離線同步任務配置指導

• 操作流程請參見嚮導模式配置、指令碼模式配置。

• 指令碼模式配置的全量參數和指令碼Demo請參見下文的附錄：指令碼Demo與參數說明。

單表、整庫即時同步任務配置指導

操作流程請參見單表即時同步任務配置、整庫即時同步任務配置。

SSL

配置Kafka資料來源時，特殊認證方式選擇SSL或者SASL_SSL時，表明Kafka叢集開啟了SSL認證，您需要上傳用戶端truststore認證檔案並填寫truststore認證密碼。

• 如果Kafka叢集為alikafka執行個體，可以參考SSL認證演算法升級說明下載正確的truststore認證檔案，truststore認證密碼為KafkaOnsClient。

• 如果Kafka叢集為EMR執行個體，可以參考使用SSL加密Kafka連結下載正確的truststore認證檔案並擷取truststore認證密碼。

• 如果是自建叢集，請自行上傳正確的truststore認證，填寫正確的truststore認證密碼。

keystore認證檔案、keystore認證密碼和SSL密鑰密碼只有在Kafka叢集開啟雙向SSL認證時才需要進行配置，用於Kafka叢集服務端認證用戶端身份。當Kafka叢集server.properties中ssl.client.auth=required時開啟雙向SSL認證，詳情請參見使用SSL加密Kafka連結。

GSSAPI

配置Kafka資料來源時，當Sasl機制選擇GSSAPI時，需要上傳三個認證檔案，分別是JAAS設定檔、Kerberos設定檔以及Keytab檔案，並在獨享資源群組進行DNS/HOST設定，下面分別介紹這三種檔案以及獨享資源群組DNS、HOST的配置方式。

• JAAS設定檔

  JAAS檔案必須以KafkaClient開頭，之後使用一個大括弧包含所有配置項：

  • 大括弧內第一行定義使用的登入組件類，對於各類Sasl認證機制，登入組件類是固定的，後續的每個配置項以key=value格式書寫。

  • 除最後一個配置項外，其他配置項結尾不能有分號。

  • 最後一個配置項結尾必須有分號，在大括弧}之後也必須加上一個分號。

  不符合以上格式要求將導致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 principal，請根據實際情況配置該項，並確保上傳的keytab檔案包含該principal的密鑰。

• Kerberos設定檔

  Kerberos設定檔必須包含兩個模組[libdefaults]和[realms]

  • [libdefaults]模組指定Kerberos認證參數，模組中每個配置項以key=value格式書寫。

  • [realms]模組指定KDC地址，可以包含多個realm子模組，每個realm子模組以realm名稱=開頭。

  後面緊跟一組用大括弧包含配置項，每個配置項也以key=value格式書寫，典型的Kerberos設定檔格式如下（根據實際情況替換以下內容中的xxx）：

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

  配置項	說明
  [libdefaults].default_realm	訪問Kafka叢集節點時預設使用的realm，一般情況下與JAAS設定檔中指定用戶端principal所在realm一致。
  [libdefaults]其他參數	[libdefaults]模組可以指定其他一些Kerberos認證參數，例如ticket_lifetime等，請根據實際需要配置。
  [realms].realm名稱	需要與JAAS設定檔中指定用戶端principal所在realm，以及[libdefaults].default_realm一致，如果JAAS設定檔中指定用戶端principal所在realm和[libdefaults].default_realm不一致，則需要包含兩組realms子模組分別對應JAAS設定檔中指定用戶端principal所在realm和[libdefaults].default_realm。
  [realms].realm名稱.kdc	以ip:port格式指定KDC地址和連接埠，例如kdc=10.0.0.1:88，連接埠如果省略預設使用88連接埠，例如kdc=10.0.0.1。

• Keytab檔案

  Keytab檔案需要包含JAAS設定檔指定principal的密鑰，並且能夠通過KDC的驗證。例如本地當前工作目錄有名為client.keytab的檔案，可以通過以下命令驗證Keytab檔案是否包含指定principal的密鑰。

  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配置

  開啟Kerberos認證的Kafka叢集，會使用Kafka叢集中節點的hostname作為節點在KDC（Kerberos的服務端程式，即密鑰分發中心）中註冊的principal的一部分，而用戶端訪問Kafka叢集節點時，會根據本地的DNS、HOST設定，推導Kafka叢集節點的principal，進而從KDC擷取節點的訪問憑證。使用獨享資源群組訪問開啟Kerberos認證的Kafka叢集時，需要正確配置DNS、HOST，以確保順利從KDC擷取Kafka叢集節點的訪問憑證：

  • DNS設定

    當獨享資源群組綁定的VPC中，使用PrivateZone執行個體進行了Kafka叢集節點的網域名稱解析設定，則可以在DataWorks管控台的獨享資源群組對應的VPC繫結項目增加100.100.2.136和100.100.2.138兩個IP的自訂路由，以使PrivateZone針對Kafka叢集節點的網域名稱解析設定對獨享資源群組生效。在 DataWorks 控制台左側導覽列單擊資源群組列表，在獨享資源群組的操作列單擊網路設定，進入專用網路綁定頁簽後在操作列單擊自訂路由，在彈窗中單擊新增路由，將目標類型選擇為IDC，串連方式選擇為直連IP，填寫直連 IP 位址，然後單擊產生路由完成配置。

  • HOST設定

    當獨享資源群組綁定的VPC中，未使用PrivateZone執行個體進行了Kafka叢集節點的網域名稱解析設定，則需要在DataWorks管控台的獨享資源群組網路設定中逐個將Kafka叢集節點的IP地址與網域名稱映射添加到Host配置中。在DataWorks管理主控台左側導覽列單擊資源群組列表，找到目標資源群組，單擊其操作列的網路設定，選擇Host配置頁簽，單擊新增按鈕添加Host網域名稱。Host配置優先順序高於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指令碼Demo

從Kafka讀取資料的JSON配置如下。

{
    "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"//限流，此處1mbps = 1MB/s。
        }
    }
}

Reader指令碼參數

kafkaConfig參數說明如下。

Writer指令碼Demo

向Kafka寫入資料的JSON配置，如下所示。

{
  "type":"job",
  "version":"2.0",//版本號碼。
  "steps":[
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType":"Kafka",//外掛程式名。
      "parameter":{
          "server": "ip:9092", //Kafka的server地址。
          "keyIndex": 0, //作為Key的列。需遵循駝峰命名規則，k小寫
          "valueIndex": 1, //作為Value的某列。目前只支援取來源端資料的一列或者該參數不填（不填表示取來源所有資料）
          //例如想取odps的第2、3、4列資料作為kafkaValue，請建立odps表將原odps表資料做清洗整合寫新odps表後使用新表同步。
          "keyType": "Integer", //Kafka的Key的類型。
          "valueType": "Short", //Kafka的Value的類型。
          "topic": "t08", //Kafka的topic。
          "batchSize": 1024 //向kafka一次性寫入的資料量，單位是位元組。
        },
      "name":"Writer",
      "category":"writer"
    }
  ],
  "setting":{
      "errorLimit":{
      "record":"0"//錯誤記錄數。
    },
    "speed":{
        "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
        "concurrent":1, //作業並發數。
        "mbps":"12"//限流，此處1mbps = 1MB/s。
    }
   },
    "order":{
    "hops":[
        {
            "from":"Reader",
            "to":"Writer"
        }
      ]
    }
}

Writer指令碼參數