отправка сообщений от потребителя кафки на mongodb - PullRequest
Новая
       18

отправка сообщений от потребителя кафки на mongodb

0 голосов
/ 04 апреля 2019

Я создал потребитель kafka, используя 'kafka-node', для файла события 

consumer.on('message' ()=>{
connecting to mongodb and inserting to a collection.
})

mongo.js, используемого для создания соединения с mongo и возврата объекта 

const MongoClient = require('mongodb').MongoClient, assert = require('assert');

const url = 'mongodb://root:****@ds031257.mlab.com:31257/kafka-node';

let _db;

 const connectDB =  (callback) => {
     try {
         MongoClient.connect(url, { useNewUrlParser: true }, (err, database) => {
             console.log('message' + database)
             _db = database.db('kafka-node');
             return callback(err);
         })
     } catch (e) {
         throw e;
     }
 }

 const getDB = () => _db;

 const close = () => _db.close();
 module.exports = { connectDB, getDB, close }

consumer.js - это создание потребителя и отправка сообщений на mongodb 

let kafka = require('kafka-node');
let MongoDB = require('./mongo');
let Consumer = kafka.Consumer,
    // The client specifies the ip of the Kafka producer and uses
    // the zookeeper port 2181
    client = new kafka.KafkaClient({ kafkaHost: 'localhost:9093, localhost:9094, localhost:9095' });
// The consumer object specifies the client and topic(s) it subscribes to
consumer = new Consumer(
    client, [{ topic: 'infraTopic', partitions: 3 }], { autoCommit: false });


consumer.on('ready', function () {
    console.log('consumer is ready');
});

consumer.on('error', function (err) {
    console.log('consumer is in error state');
    console.log(err);
})
client.refreshMetadata(['infraTopic'], (err) => {
    if (err) {
        console.warn('Error refreshing kafka metadata', err);
    }
});
consumer.on('message', function (message) {
    // grab the main content from the Kafka message
    console.log(message);
    MongoDB.connectDB((err) => {
        if (err) throw err
        // Load db & collections
        const db = MongoDB.getDB();
        const collectionKafka = db.collection('sampleCollection');
        try {
            collectionKafka.insertOne(
                {
                    timestamp: message.value,
                    topic: message.topic
                },
                function (err, res) {
                    if (err) {
                        database.close();
                        return console.log(err);
                    }
                    // Success
                }
            )
        } catch (e) {
            throw e
        }
    })
});

. Это правильный способ отправки сообщений на mongodb от потребителя kafka?с этой настройкой он работает до тех пор, пока все сообщения не будут записаны, и как только он достигнет EOL, он выбрасывает "Не удается прочитать свойство 'db' из null"

1 Ответ

1 голос
/ 04 апреля 2019

это правильный способ отправлять сообщения на mongodb от потребителя kafka?

Полагаю, это один путь, но я бы не назвал его "1007 * правильный путь" :)

Намного лучше использовать Kafka Connect. Он является частью Apache Kafka и предназначен для того, чтобы делать именно то, что вы пытаетесь сделать - передавать данные из Kafka в целевую систему (вы также можете использовать его для потоковой передачи данных из других систем в Kafka) ,

Для MongoDB имеется отличный превосходный разъем с исчерпывающей документацией , который будет делать именно то, что вы пытаетесь сделать.

Если вам нужно обработать данные перед их записью, то шаблон, который следует использовать, обрабатывается с помощью Kafka Streams, KSQL или любого другого инструмента обработки, который вы хотите использовать, но запишите его обратно в тему Kafka . Затем эта тема читается Kafka Connect и направляется к вашей цели. Таким образом, вы разделяете обязанности и создаете гораздо более простую, но при этом гибкую и масштабируемую систему.

...