Kafka作为分布式消息队列,广泛应用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域,您可以使用Kafka数据源配置同步任务同步数据。本文为您介绍如何创建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_PLAINTEXTSASL_SSLSSL),开启第三方认证功能后,只有可信的应用和服务才能访问数据资源。
      说明
      • 使用第三方认证功能时,您需要提前在DataWorks的认证文件管理页面上传认证文件。上传和引用认证文件,详情请参见上传和引用认证文件
      • 如果访问数据源时您无需对其他应用或服务进行认证,则可将特殊认证方式配置为None
      • 当前仅西南1(成都)地域支持Kafka数据源配置第三方认证功能。
      • 阿里云实例模式连接串模式类型的数据源,认证方式的配置相同。
      不同认证方式的配置如下:
      • 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)时,需要配置该参数。
        Jaas配置文件
        该文件包含认证和授权信息,用于设置Java的安全鉴权系统参数java.security.auth.login.config您可以选择DataWorks中已上传的文件,也可以单击新增认证文件,上传新的认证文件。
        说明Sasl机制参数配置为GSSAPI(Kerberos)时,数据集成功能在使用Jaas配置文件时会引用Keytab文件的秘钥信息进行认证。
      • 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)颁发的数字证书,用于验证访问源可信任性。
        Truststore密码

        读取Kafka集群CA证书内容时使用的密码。

        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)时,需要配置该参数。
        Jaas配置文件
        该文件包含认证和授权信息,用于设置Java的安全鉴权系统参数java.security.auth.login.config您可以选择DataWorks中已上传的文件,也可以单击新增认证文件,上传新的认证文件。
        说明Sasl机制参数配置为GSSAPI(Kerberos)时,数据集成功能在使用Jaas配置文件时会引用Keytab文件的秘钥信息进行认证。
      • SSL主要用于客户端到服务器端的认证。使用SSL方式进行认证的配置如下。SSL
        参数 描述
        Truststore证书文件
        存放Kafka集群CA证书的文件,用于访问SSL服务器时对该证书进行认证,以确保访问本数据源的应用或服务是可信任的。您可以选择DataWorks中已上传的文件,也可以单击新增认证文件,上传新的认证文件。
        说明 CA证书是由电子商务认证中心CA(Certificate Authority)颁发的数字证书,用于验证访问源可信任性。
        Truststore密码

        读取Kafka集群CA证书内容时使用的密码。

        Keystore 证书文件

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

        Keystore 密码

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

        SSL秘钥密码

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

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

后续步骤

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