Добавить точку сохранения в Datastream

MongoDB не разрешает объединения, но вы можете использовать плагины для его обработки. Проверьте плагин mongo-join. Это лучшее, и я уже использовал его. Вы можете установить его с помощью npm прямо следующим образом npm install mongo-join. Вы можете проверить полную документацию с примерами .

(++) действительно полезный инструмент, когда нам нужно объединить (N) коллекции

(- ), мы можем применять условия только на верхнем уровне запроса

Пример

var Join = require('mongo-join').Join, mongodb = require('mongodb'), Db = mongodb.Db, Server = mongodb.Server;
db.open(function (err, Database) {
    Database.collection('Appoint', function (err, Appoints) {

        /* we can put conditions just on the top level */
        Appoints.find({_id_Doctor: id_doctor ,full_date :{ $gte: start_date },
            full_date :{ $lte: end_date }}, function (err, cursor) {
            var join = new Join(Database).on({
                field: '_id_Doctor', // <- field in Appoints document
                to: '_id',         // <- field in User doc. treated as ObjectID automatically.
                from: 'User'  // <- collection name for User doc
            }).on({
                field: '_id_Patient', // <- field in Appoints doc
                to: '_id',         // <- field in User doc. treated as ObjectID automatically.
                from: 'User'  // <- collection name for User doc
            })
            join.toArray(cursor, function (err, joinedDocs) {

                /* do what ever you want here */
                /* you can fetch the table and apply your own conditions */
                .....
                .....
                .....


                resp.status(200);
                resp.json({
                    "status": 200,
                    "message": "success",
                    "Appoints_Range": joinedDocs,


                });
                return resp;


            });

    });

0
задан Ahmad Qureshi 17 January 2019 в 14:45
поделиться

1 ответ

Несколько уточнений ...

  1. Даже если вы поддерживаете нижестоящее состояние, источник Кафки должен поддерживать какое-то состояние, чтобы знать, какую часть темы он занимал; при перезапуске с контрольной точки или точки сохранения он должен воспроизвести любые данные со времени последних данных, которые перешли в сохраненное состояние, и текущих доступных данных.
  2. Любое состояние, которое вы сохраняете, должно иметь некоторую продолжительность (обычно ограничение по времени), чтобы оно не накапливалось неограниченно, верно?

Самый простой подход, который я могу себе представить это превращение вашего метода App.convertToObject() в правильную функцию, которая реализует интерфейс [ListCheckpointed][1]. Вы сохраняете каждый преобразованный объект в список до определенного предела (см. Пример BufferingSink на той же странице).

0
ответ дан kkrugler 17 January 2019 в 14:45
поделиться
Другие вопросы по тегам:

Похожие вопросы: