:Utiliser un client Apache Kafka open source pour écrire des données dans le moteur de streaming Lindorm

Prérequis

• Un environnement Java installé via le JDK (Java Development Kit) 1.7 ou une version ultérieure.

• Ajoutez l'adresse IP de votre client à la liste d'autorisation de votre instance Lindorm. Pour plus d'informations, consultez Configurer une liste d'autorisation.

• Récupérez la valeur de l'endpoint Lindorm Stream Kafka. Pour plus d'informations, consultez Consulter les endpoints.

  Remarque

  L'endpoint Lindorm Stream Kafka correspond à un endpoint VPC (Virtual Private Cloud) de votre moteur de streaming Lindorm. Assurez-vous que votre application et votre instance Lindorm sont déployées dans le même VPC.

Procédure

1. Téléchargez un client Apache Kafka open source. Ajoutez les dépendances Maven au fichier pom.xml. Voici un exemple :

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

2. Connectez-vous au moteur de streaming Lindorm et écrivez-y des données. Le code complet est fourni ci-dessous :

   Remarque

   • Le moteur de streaming Lindorm accepte les données aux formats JSON, Avro ou CSV.

   • Dans l'exemple de code, la valeur de l'endpoint Lindorm Stream Kafka correspond à un endpoint VPC. Pour savoir comment obtenir cet endpoint, consultez Consulter les endpoints.

   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();
   
           // Configure Lindorm Stream Kafka Endpoint. The value of Lindorm Stream Kafka Endpoint is a VPC endpoint. Make sure that your application and your Lindorm instance are deployed in the same VPC. 
           props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "Lindorm Stream Kafka Endpoint");
          // Specify the topic in which you want to store the physical data of your streaming data table.
           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();
               // Write data to the streaming engine.
               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();
           }
       }
   }