Import data from Kafka¶

{
  # Spark configuration
  spark: {
    app: {
      name: Nebula Exchange 2.6.0
    }
    driver: {
      cores: 1
      maxResultSize: 1G
    }
    cores {
      max: 16
    }
  }


  # Nebula Graph configuration
  nebula: {
    address:{
      # Specify the IP addresses and ports for Graph and all Meta services.
      # If there are multiple addresses, the format is "ip1:port","ip2:port","ip3:port".
      # Addresses are separated by commas.
      graph:["127.0.0.1:9669"]
      meta:["127.0.0.1:9559"]
    }
    # The account entered must have write permission for the Nebula Graph space.
    user: root
    pswd: nebula
    # Fill in the name of the graph space you want to write data to in the Nebula Graph.
    space: basketballplayer
    connection {
      timeout: 3000
      retry: 3
    }
    execution {
      retry: 3
    }
    error: {
      max: 32
      output: /tmp/errors
    }
    rate: {
      limit: 1024
      timeout: 1000
    }
  }
  # Processing vertexes
  tags: [
    # Set the information about the Tag player.
    {

      # The corresponding Tag name in Nebula Graph.
      name: player
      type: {
        # Specify the data source file format to Kafka.
        source: kafka
        # Specify how to import the data into Nebula Graph: Client or SST.
        sink: client
      }
      # Kafka server address.
      service: "127.0.0.1:9092"
      # Message category.
      topic: "topic_name1"

      # Kafka data has a fixed domain name: key, value, topic, partition, offset, timestamp, timestampType.
      # If multiple fields need to be specified after Spark reads as DataFrame, separate them with commas.
      # Specify the field name in fields. For example, use key for name in Nebula and value for age in Nebula, as shown in the following.
      fields: [key,value]
      nebula.fields: [name,age]

      # Specify a column of data in the table as the source of vertex VID in the Nebula Graph.
      # The key is the same as the value above, indicating that key is used as both VID and property name.
      vertex:{
          field:key
      }


      # The number of data written to Nebula Graph in a single batch.
      batch: 10

      # The number of Spark partitions.
      partition: 10
      # The interval for message reading. Unit: second.
      interval.seconds: 10
    }
    # Set the information about the Tag Team.
    {
      name: team
      type: {
        source: kafka
        sink: client
      }
      service: "127.0.0.1:9092"
      topic: "topic_name2"
      fields: [key]
      nebula.fields: [name]
      vertex:{
          field:key
      }
      batch: 10
      partition: 10
      interval.seconds: 10
    }

  ]

  # Processing edges
  edges: [
    # Set the information about the Edge Type follow.
    {
      # The corresponding Edge Type name in Nebula Graph.
      name: follow

      type: {
        # Specify the data source file format to Kafka.
        source: kafka

        # Specify how to import the Edge type data into Nebula Graph.
        # Specify how to import the data into Nebula Graph: Client or SST.
        sink: client
      }

      # Kafka server address.
      service: "127.0.0.1:9092"
      # Message category.
      topic: "topic_name3"

      # Kafka data has a fixed domain name: key, value, topic, partition, offset, timestamp, timestampType.
      # If multiple fields need to be specified after Spark reads as DataFrame, separate them with commas.
      # Specify the field name in fields. For example, use key for degree in Nebula, as shown in the following.
      fields: [key]
      nebula.fields: [degree]

      # In source, use a column in the topic as the source of the edge's source vertex.
      # In target, use a column in the topic as the source of the edge's destination vertex.
      source:{
          field:timestamp
      }


      target:{
          field:offset
      }

      # (Optional) Specify a column as the source of the rank.
      #ranking: rank

      # The number of data written to Nebula Graph in a single batch.
      batch: 10

      # The number of Spark partitions.
      partition: 10

      # The interval for message reading. Unit: second.
      interval.seconds: 10
    }

    # Set the information about the Edge Type serve.
    {
      name: serve
      type: {
        source: kafka
        sink: client
      }
      service: "127.0.0.1:9092"
      topic: "topic_name4"

      fields: [timestamp,offset]
      nebula.fields: [start_year,end_year]
      source:{
          field:key
      }

      target:{
          field:value
      }

      # (Optional) Specify a column as the source of the rank.
      #ranking: rank

      batch: 10
      partition: 10
      interval.seconds: 10
    }
  ]
}

${SPARK_HOME}/bin/spark-submit --master "local" --class com.vesoft.nebula.exchange.Exchange <nebula-exchange-2.6.0.jar_path> -c <kafka_application.conf_path>

${SPARK_HOME}/bin/spark-submit  --master "local" --class com.vesoft.nebula.exchange.Exchange  /root/nebula-exchange/nebula-exchange/target/nebula-exchange-2.6.0.jar  -c /root/nebula-exchange/nebula-exchange/target/classes/kafka_application.conf

GO FROM "player100" OVER follow;