我们如何连接AMPS[CRANK UP THE AMPS]Server和Apache Flink以实现实时流

Question

我们从AMPS[CRANK UP the AMPS]服务器订阅实时数据，作为Apache flink的来源。任何关于如何像卡夫卡一样将两者联系起来的想法。

Amps服务器：http://www.crankuptheamps.com/amps/

Answer 1

目前，Apache Flink没有为AMPS提供任何开箱即用的连接器，正如您在这里看到的那样。但是，它确实提供了一个可扩展的源/汇接口，可以用于接入任何自定义源/汇。

您可以通过扩展RichSourceFunction并将其传递给本flink文档中提到的addSource方法来创建自己的AMPS源连接器。请参阅crankupthemps提供的Java客户端API，以连接到源主题并订阅消息。

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import com.crankuptheamps.client.Client;
import com.crankuptheamps.client.Message;
public class AMPSSource extends RichSourceFunction<String> {

private static final long serialVersionUID = -8708182052610791593L;
private String name, topic, connectionString;
private Client client;
public AMPSSource(String name, String connectionString, String topic) {
this.name = name;
this.topic = topic;
this.connectionString = connectionString;
}
@Override
public void open(Configuration parameters) throws Exception {
// We create a Client, then connect() and logon()
client = new Client(this.name);
client.connect(this.connectionString);
client.logon();
}
public void run(SourceContext<String> sourceContext) throws Exception {
/*
* Here, we iterate over messages in the MessageStream returned by
* subscribe method
*/
for (Message message : client.subscribe(this.topic)) {
sourceContext.collect(message.getData());
}
}
@Override
public void close() throws Exception {
try {
cancel();
} finally {
super.close();
}
}
public void cancel() {
client.close();
}
}

这可以用作处理器中的源，如下所示，

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class StreamProcessor {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> ampsStream = env
.addSource(new AMPSSource("flink-consumer", "tcp://127.0.0.1:9007/amps/json", "test-topic"));
ampsStream.print();
env.execute();
}
}

注意：RichSourceFunction实现的并行度为1。要启用并行执行，用户定义的源应该实现org.apache.flink.streaming.api.functions.source.ParallelSourceFunction或扩展org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction