Kafka作为分布式消息队列，广泛应用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域，您可以使用Kafka数据源配置同步任务同步数据。本文为您介绍如何创建Kafka数据源。

背景信息

标准模式的工作空间支持数据源隔离功能，您可以分别添加并隔离开发环境和生产环境的数据源，以保护您的数据安全。详情请参见数据源开发和生产环境隔离。

注意事项

支持阿里云Kafka，以及>=0.10.2且<=2.2.x的自建Kafka版本。
说明 对于<0.10.2版本Kafka，由于Kafka不支持检索分区数据offset，并且Kafka数据结构可能不支持时间戳，进而无法支持数据同步。

新增数据源

  1. 进入数据源管理页面。
    1. 登录DataWorks控制台。
    2. 在左侧导航栏，单击工作空间列表。
    3. 选择工作空间所在地域后，单击相应工作空间后的数据集成。
    4. 在左侧导航栏，单击数据源，进入数据源管理页面。
  2. 在数据源管理页面，单击右上角的新增数据源。
  3. 在新增数据源对话框中，选择数据源类型为Kafka。
  4. 在新增Kafka数据源对话框中，配置各项参数。
    1. 配置数据源的基本信息。
      Kafka数据源包括阿里云实例模式和连接串模式两种类型。
      • 以新增阿里云实例模式类型的数据源为例，配置数据源的基本信息。Kafka阿里云实例模式
        参数描述
        数据源类型当前选择的数据源类型为阿里云实例模式。
        数据源名称

        数据源名称必须以字母、数字、下划线（_）组合，且不能以数字和下划线（_）开头。

        数据源描述

        对数据源进行简单描述，不得超过80个字符。

        适用环境
        可以选择开发或生产环境。
        说明 仅标准模式工作空间会显示该配置。
        地区选择需要同步的Kafka实例所在的地域。
        实例ID输入需要同步的Kafka实例ID。您可以登录Kafka管控台，进入实例列表页面，获取实例ID。
      • 以新增连接串模式类型的数据源为例，配置数据源的基本信息。Kafka连接串模式
        参数描述
        数据源类型当前选择的数据源类型为连接串模式。
        数据源名称

        数据源名称必须以字母、数字、下划线（_）组合，且不能以数字和下划线（_）开头。

        数据源描述

        对数据源进行简单描述，不得超过80个字符。

        适用环境
        可以选择开发或生产环境。
        说明 仅标准模式工作空间会显示该配置。
        Kafka集群地址Kafka集群Broker的IP地址和端口，即Kafka实例的接入点信息。您可以登录Kafka管控台，进入实例列表页面，单击实例名称，在实例详情页获取接入点信息。

        多个Broker地址之间使用逗号（,）分隔，示例为10.0.0.1:9092,10.0.0.2:9092。

    2. 配置数据源的认证信息。
      第三方认证机制用于用户和服务的强身份验证，通过该机制，可以有效的避免不受信任的程序或服务来获取数据访问权限，提高数据同步过程中访问数据资源的安全性。DataWorks在配置Kafka数据源时为您提供了三种第三方认证方式（即特殊认证方式配置为SASL_PLAINTEXT、SASL_SSL或SSL），开启第三方认证功能后，只有可信的应用和服务才能访问数据资源。
      说明
      • 使用第三方认证功能时，您需要提前在DataWorks的认证文件管理页面上传认证文件。上传和引用认证文件，详情请参见配置第三方身份认证。
      • 如果访问数据源时您无需对其他应用或服务进行认证，则可将特殊认证方式配置为None。
      • 阿里云实例模式和连接串模式类型的数据源，认证方式的配置相同。
      不同认证方式的配置如下：
      • SASL_PLAINTEXT是一种简单的用户名和密码认证机制。使用SASL_PLAINTEXT方式进行认证的配置如下。SASL_PLAINTEXT
        参数描述
        Sasl机制

        目前支持使用GSSAPI(Kerberos)和PLAIN（指通过用户名和密码认证）两种简单验证和安全层Sasl（Simple Authentication and Security Layer）认证机制。

        Keytab文件
        密钥表文件，用于存储其他应用或服务的密钥信息，后续在Jaas配置文件中会引用该文件对其他应用或服务进行身份验证。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明 仅当Sasl机制参数配置为GSSAPI(Kerberos)时，需要配置该参数。
        Kerberos配置文件
        该文件包含密钥分发中心KDC（Key Distribute Center）的地址信息，用于设置Java的安全认证系统参数java.security.krb5.conf。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • 仅当Sasl机制参数配置为GSSAPI(Kerberos)时，需要配置该参数。
        • 典型的Kerberos配置文件格式如下：
           [libdefaults]
    kdc_realm = EMR.232619.COM
    default_realm = EMR.232619.COM
    udp_preference_limit = 1
    kdc_tcp_port = 88
    kdc_udp_port = 88
    dns_lookup_kdc=false

[realms]
    EMR.232619.COM = {
                kdc = 106.15.184.100:88
    }
        Jaas配置文件
        该文件包含认证和授权信息，用于设置Java的安全鉴权系统参数java.security.auth.login.config。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • 当Sasl机制参数配置为GSSAPI(Kerberos)时，数据集成功能在使用Jaas配置文件时会引用Keytab文件的密钥信息进行认证。
        • 典型的JAAS配置文件格式如下：
            KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="Lisa"
  password="secret";
};
注意JaasContextName务必指定为KafkaClient。
      • SASL_SSL是一种主要用于客户端到服务器端进行简单认证的认证方式。使用SASL_SSL方式进行认证的配置如下。SASL_SSL
        参数描述
        Sasl机制

        目前支持使用GSSAPI(Kerberos)和PLAIN（指通过用户名和密码认证）两种简单验证和安全层Sasl（Simple Authentication and Security Layer）认证机制。

        Truststore证书文件
        存放Kafka集群CA证书的文件，用于访问SSL服务器时对该证书进行认证，以确保访问本数据源的应用或服务是可信任的。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • CA证书是由电子商务认证中心CA（Certificate Authority）颁发的数字证书，用于验证访问源可信任性。
        • 如果您使用的是阿里kafka实例，请参考SSL证书算法升级说明获取证书。
        Truststore密码
        读取Kafka集群CA证书内容时使用的密码。
        说明 如果您使用的是阿里kafka实例，密码固定为KafkaOnsClient。
        Keystore 证书文件

        存放Kafka集群可信的CA证书和密钥的密钥库文件。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。

        Keystore 密码

        读取Keystore 证书文件内容的密码。

        SSL密钥密码

        读取Keystore 证书文件中指定密钥对的密码。

        Keytab文件
        密钥表文件，用于存储其他应用或服务的密钥信息，后续在Jaas配置文件中会引用该文件对其他应用或服务进行身份验证。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明 仅当Sasl机制参数配置为GSSAPI(Kerberos)时，需要配置该参数。
        Kerberos配置文件
        该文件包含密钥分发中心KDC（Key Distribute Center）的地址信息，用于设置Java的安全认证系统参数java.security.krb5.conf。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • 仅当Sasl机制参数配置为GSSAPI(Kerberos)时，需要配置该参数。
        • 典型的Kerberos配置文件格式如下：
           [libdefaults]
    kdc_realm = EMR.232619.COM
    default_realm = EMR.232619.COM
    udp_preference_limit = 1
    kdc_tcp_port = 88
    kdc_udp_port = 88
    dns_lookup_kdc=false

[realms]
    EMR.232619.COM = {
                kdc = 106.15.184.100:88
    }
        Jaas配置文件
        该文件包含认证和授权信息，用于设置Java的安全鉴权系统参数java.security.auth.login.config。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • 当Sasl机制参数配置为GSSAPI(Kerberos)时，数据集成功能在使用Jaas配置文件时会引用Keytab文件的密钥信息进行认证。
        • 典型的JAAS配置文件格式如下：
            KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="Lisa"
  password="secret";
};
注意JaasContextName务必指定为KafkaClient。
      • SSL主要用于客户端到服务器端的认证。使用SSL方式进行认证的配置如下。SSL
        参数描述
        Truststore证书文件
        存放Kafka集群CA证书的文件，用于访问SSL服务器时对该证书进行认证，以确保访问本数据源的应用或服务是可信任的。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。
        说明
        • CA证书是由电子商务认证中心CA（Certificate Authority）颁发的数字证书，用于验证访问源可信任性。
        • 如果您使用的是阿里kafka实例，请参考SSL证书算法升级说明获取证书。
        Truststore密码
        读取Kafka集群CA证书内容时使用的密码。
        说明 如果您使用的是阿里kafka实例，密码固定为KafkaOnsClient。
        Keystore 证书文件

        存放Kafka集群可信的CA证书和密钥的密钥库文件。您可以选择DataWorks中已上传的文件，也可以单击新增认证文件，上传新的认证文件。

        Keystore 密码

        读取Keystore 证书文件内容的密码。

        SSL密钥密码

        读取Keystore 证书文件中指定密钥对的密码。

  5. 可选：配置数据源的扩展参数。
    您可以选择为当前数据源配置扩展参数，即配置Kafka消费者和生产者的相关参数，格式为JSON格式。扩展参数
    示例如下：
    • 配置发往每个分区（Partition）的消息缓存量（消息内容的字节数总和）为16342。
    • 配置每条消息在缓存中的最长时间为10毫秒。
    {
"batch.size":"16342",
"linger.ms":"10"
}
    配置项可选的压缩类型有：gzip、snappy、lz4。
    {
"compression.type":"gzip"
}
    说明 如果使用脚本模式配置的离线同步任务，或使用单表同步配置的实时同步任务中，配置的消费者或生产者参数，与扩展参数中配置的参数相同但取值不同，则扩展参数配置的参数优先级低于同步任务中配置的参数。
  6. 测试数据源与资源组的连通性。
    1. 选择资源组连通性类型为数据集成。
    2. 在资源组列表，单击相应资源组后的测试连通性。
      数据同步时，一个任务只能使用一种资源组。您需要测试每个资源组的连通性，以保证同步任务使用的数据集成资源组能够与数据源连通，否则将无法正常执行数据同步任务。如果您需要同时测试多种资源组，请选中相应资源组后，单击批量测试连通性。详情请参见配置资源组与网络连通。
      说明
      • （推荐）资源组列表默认仅显示独享数据集成资源组，为确保数据同步的稳定性和性能要求，推荐使用独享数据集成资源组。
      • 如果您需要测试公共资源组或自定义资源组的连通性，请在资源组列表右下方，单击更多选项，在警告对话框单击确定，资源组列表会显示可供选择的公共资源组和自定义资源组。
  7. 测试连通性通过后，单击完成。

后续步骤

您可以使用创建的数据源，执行数据集成同步任务，详情请参见数据集成概述。