Kafka 配置

本文介绍从 Kafka 数据源导入数据的配置项。关于完整配置文件和配置项的信息，请参见数据导入配置。

数据源配置

下表列出了 Kafka 数据源特定的配置项：

选项	描述	类型	是否必选	默认值
type	数据源的类型。设置为"kafka"。	String	是	无
server	Kafka 服务器的地址和端口。	String	是	无
topic	从中检索数据的 Kafka 主题。	String	是	无
startingOffsets	从 Kafka 主题读取数据的起始偏移量。 有效值为"earliest"和"latest"。从最早的偏移量读取数据，请设置为"earliest"。从最新的偏移量读取数据，请设置为"latest"。	String	否	"earliest"
maxOffsetsPerTrigger	每个触发间隔处理的最大偏移量数。	Int	否	null
securityProtocol	用于与 Kafka 服务器通信的安全协议。有效值为"PLAINTEXT"、"SASL_PLAINTEXT"、"SASL_SSL"和"SSL"。 此选项仅在 Kafka 服务器配置了身份验证时必选。	String	否	null
mechanism	用于身份验证的安全机制。有效值为"GSSAPI"、"PLAIN"、"SCRAM-SHA-256"和"SCRAM-SHA-512"。 此选项仅在 Kafka 服务器配置了身份验证时必选。	String	null	无
kerberos	是否启用 Kerberos 身份验证。 当 Kafka 服务器使用 Kerberos 进行身份验证时，此选项必须设置为true。	Bool	否	false
kerberosServiceName	Kerberos 服务名称。 此选项仅在启用 Kerberos 身份验证时必选。	String	否	null

通用源配置

下列配置项也可以作为全局设置在配置文件的sourcesCommon一节指定：

• type
• server
• startingOffsets
• maxOffsetsPerTrigger
• securityProtocol
• mechanism
• kerberos
• kerberosServiceName

请注意，针对特定数据源的配置会覆盖全局设置。

数据源配置示例

Kafka 源配置

sources: [
  {
      type:"kafka"
      server: "127.0.0.1:9092"
      topic:"movie"
      startingOffsets: "earliest"
      maxOffsetsPerTrigger: 1000000
      # if your kafka server is authenticated, then you need to config the following four configs.
      securityProtocol: SASL_PLAINTEXT
      mechanism: GSSAPI
      kerberos: true
      kerberosServiceName: kafka
//      nodetypes: []
//      edgetypes: []
  }
]    

完整配置示例

将<current-password>替换为登录账号的当前密码。若已修改密码，请使用修改后的密码。

{
  taskType: import
  nebula: {
    graphAddr: "127.0.0.1:9669"
    graphName: "movie"
    user: "root"
    passwd: "<current-password>"
  }
  error: {
    path: "/tmp/errors"
    maxRecords: 10000
  }
  sourcesCommon:{
    type:"kafka"
    server: "127.0.0.1:9092"
    startingOffsets: "earliest"
    maxOffsetsPerTrigger: 1000000
  }
  writeCommon:{
    batchSize: 2000
    enableRealTimeQPSLog: true
    mode: "INSERT IGNORE"
  }
  sources: [
    # node type Actor
    {
      topic: actor
      pre_processes: {
          filter: ["birth_date!='\\N'"]
          concatConstant: [
            {
              field: "name"
              prefixValue: "test_"
              suffixValue: ""
            }
          ]
      }
      nodetypes: [
        {
          name: "Actor"
          sourceFields: [people_id,name,birth_date]
          nebulaFields: [id,name,birthDate]
        }
      ]
    }
    # node type Movie
    {
      topic: movie
      nodetypes: [
        {
          name: "Movie"
          sourceFields: [movie_id,name]
          nebulaFields: [id,name]
        }
      ]
    }
   # edge type Act
    {
      topic: act
      edgetypes: [
        {
          name: "Act"
          sourceKeys: [people_id]
          nebulaSrcKeys: [id]
          targetKeys: [movie_id]
          nebulaDstKeys: [id]
          sourceFields: []
          nebulaFields: []
        }
      ]
    }

  ]
}