Как использовать источник потока Twitter в Hazelcast Jet без DAG?

Question

Я хочу провести простой анализ живого потока твитов.

Как использовать источник потока Twitter в Hazelcast Jet без DAG?

Подробнее

Инкапсуляция API Twitter довольно хороша на StreamTwitterP.java .

Однако вызывающая сторона использует это как часть DAG, c / o:

Vertex twitterSource = 
  dag.newVertex("twitter", StreamTwitterP.streamTwitterP(properties, terms));

Мой вариант использования не нуждается в мощности DAG, поэтому я бы предпочел избежать этой ненужной дополнительной сложности.

Чтобы избежать DAG, я хочу использовать SourceBuilder для определенияновый источник данных для живого потока твитов.

Я предполагаю, что будет иметь код, похожий на StreamTwitterP.java, упомянутый выше, однако мне не ясно, подходит ли использование API Hazelcast JET.

Я имел в виду пример SourceBuilder из документов .

Answer 1

Вы можете преобразовать процессор в источник конвейера:

Pipeline p = Pipeline.create();
p.drawFrom(Sources.<String>streamFromProcessor("twitter", 
    streamTwitterP(properties, terms)))
...

Есть также версия twitterSource, которая использует SourceBuilder здесь .