Kafka configurations

This topic introduces the configuration options for importing data from a Kafka data source. For information about the complete configuration file and options, see Data import configurations.

Data source configurations

The following table lists the configuration options specific to the Kafka data source.

Option	Description	Type	Required	Default value
type	The type of the data source. In this case, set the value to "kafka".	String	Yes	N/A
server	The address and port of the Kafka server.	String	Yes	N/A
topic	The Kafka topic from which data is retrieved.	String	Yes	N/A
startingOffsets	The starting offset for reading data from the Kafka topic. Valid values are "earliest" and "latest". To read data from the earliest offset, set the value to "earliest". To read data from the latest offset, set the value to "latest".	String	No	"earliest"
maxOffsetsPerTrigger	The maximum number of offsets to process per trigger interval.	Int	No	null
securityProtocol	The security protocol used to communicate with the Kafka server. Valid values are "PLAINTEXT", "SASL_PLAINTEXT", "SASL_SSL", and "SSL". This option is required only when the Kafka server is authenticated.	String	No	null
mechanism	The security mechanism used for authentication. Valid values are "GSSAPI", "PLAIN", "SCRAM-SHA-256", and "SCRAM-SHA-512". This option is required only when the Kafka server is authenticated.	String	No	null
kerberos	Whether Kerberos authentication is enabled. This option must be set to true when the Kafka server is authenticated with Kerberos.	Bool	No	false
kerberosServiceName	The Kerberos service name. This option is required only when Kerberos authentication is enabled.	String	No	null

Common source configurations

The following configuration options can also be specified as global settings in the section sourcesCommon of the configuration file:

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

Note that the configurations on specific data sources override the global settings.

Data source configuration example

Kafka source configurations

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: []
  }
]    

Complete configuration example

Replace <current-password> with the current password for the specified account. If the password has changed, use the new 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: []
        }
      ]
    }

  ]
}