通过开源Kafka客户端写入Lindorm流引擎数据

操作步骤

1. 下载开源Kafka客户端。在pom.xml中配置Maven依赖，具体内容如下：

   <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
       <version>0.10.2.2</version>
   </dependency>

2. 连接Lindorm流引擎并写入Lindorm流引擎数据。完整的代码示例如下：

   说明

   • 写入的数据格式支持JSON、Avro和CSV。

   • 代码中Lindorm Stream Kafka地址为专有网络地址，获取方法请参见查看连接地址。

   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerConfig;
   import org.apache.kafka.clients.producer.ProducerRecord;
   import org.apache.kafka.clients.producer.RecordMetadata;
   import org.codehaus.jettison.json.JSONObject;
   
   import java.util.Properties;
   import java.util.concurrent.Future;
   
   public class KafkaToLindormStreamDemo {
   
       public static void main(String[] args) {
           Properties props = new Properties();
   
           //设置Lindorm Stream Kafka地址，这个Lindorm Stream Kafka地址为专有网络地址，需确保应用程序部署的环境和Lindorm实例使用相同的专有网络ID。
           props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "Lindorm Stream Kafka地址");
          //指定数据流表的物理数据存储在某个Topic上
           String topic = "log_topic";
   
           props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
           props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
           KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
           try {
               JSONObject json = new JSONObject();
               //写入流引擎数据
               json.put("timestamp", System.currentTimeMillis());
               json.put("loglevel", "ERROR");
               json.put("thread", "[ReportFinishedTask7-thread-4]");
               json.put("class", "engine.ImporterTaskManager(318)");
               json.put("detail", "Remove tasks fail: job name=e35318e5-52ea-48ab-ad2a-0144ffc6955e , task name=prepare_e35318e5-52ea-48ab-ad2a-0144ffc6955e , runningTasks=0");
               Future<RecordMetadata> future = producer.send(
             new ProducerRecord<String, String>(topic, json.getString("thread") + json.getLong("timestamp"),
                 json.toString()));
               producer.flush();
               try {
                   RecordMetadata recordMetadata = future.get();
                   System.out.println("Produce ok:" + recordMetadata.toString());
               } catch (Throwable t) {
                   System.out.println("Produce exception " + t.getMessage());
                   t.printStackTrace();
               }
           } catch (Exception e) {
               System.out.println("Produce exception " + e.getMessage());
               e.printStackTrace();
           }
       }
   }