使用kafka实现oracle到mongodb的数据迁移

Question

我正在尝试使用Kafka将数据从oracle迁移到mongodb。我取了一个1000万的样本记录集，列长为90，每行为5KB

我将数据分成10个线程，但其中一个线程不是每次都在运行....当我检查数据时，我看到MongoDB中丢失了100万条记录。

主类：

    int totalRec = countNoOfRecordsToBeProcessed;
    int minRownum =0;
    int maxRownum =0;
    int recInThread=totalRec/10;
    System.out.println("oracle  "+new Date());
    for(int i=minRownum;i<=totalRec;i=i+recInThread+1){ 
          KafkaThread kth = new KafkaThread(i, i+recInThread, conn);
        Thread th = new Thread(kth);
        th.start();
    }
    System.out.println("oracle done+  "+new Date());    

Kafka制作人线程类：

JSONObject obj=new JSONObject（）；
while（rs.next（））{

        int total_rows = rs.getMetaData().getColumnCount();
        for (int i = 0; i < total_rows; i++) {
            obj.put(rs.getMetaData().getColumnLabel(i + 1)
                    .toLowerCase(), rs.getObject(i + 1));
     }
            //System.out.println("object->"+serializedObject);
            producer.send(new ProducerRecord<String, String>("oracle_1",obj.toString()));
            obj= new JSONObject();
        //System.out.println(counter++);
}

消费者类：

        KafkaConsumer consumer = new KafkaConsumer<>(props);
        //subscribe to topic
        consumer.subscribe(Arrays.asList(topicName));

        MongoClientURI clientURI = new MongoClientURI(mongoURI);
        MongoClient mongoClient = new MongoClient(clientURI);
        MongoDatabase database = mongoClient.getDatabase(clientURI.getDatabase());
        final MongoCollection<Document> collection = database.getCollection(clientURI.getCollection());
        while (true) {
            final ConsumerRecords<Long, String> consumerRecords =
                    consumer.poll(10000);

            if (consumerRecords.count()!=0) {
                List<InsertOneModel> list1 = new ArrayList<>();
                consumerRecords.forEach(record -> {
//                    System.out.printf("Consumer Record:(%d, %s, %d, %d)\n",
//                            record.key(), record.value(),
//                            record.partition(), record.offset());'

                    String row =null;
                     row = record.value();
                     Document doc=Document.parse(row);
                     InsertOneModel t = new InsertOneModel<>(doc);
                     list1.add(t);
                });
                collection.bulkWrite((List<? extends WriteModel<? extends Document>>) (list1), new BulkWriteOptions().ordered(false));
                consumer.commitAsync();
                list1.clear();
            }
            }
    }

Answer 1

我的建议是：使用Kafka Connect JDBC连接器将数据拉入，使用一个Kafka Connect MongoDB接收器将数据推出。否则你只是在推倒重来。卡夫卡连接是阿帕奇卡夫卡的一部分。

卡夫卡连接入门：

• https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-1/
• https://www.confluent.io/blog/blogthe-simplest-useful-kafka-connect-data-pipeline-in-the-world-or-thereabouts-part-2/
• https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-3/