实时计算框架梳理(flink,kafka,hbase/redis,clickhouse)
实时计算框架对比和示例代码:
Apache Flink
Flink 是一个分布式流处理和批处理系统。以下是使用Flink读取Kafka数据并写入HBase的简单示例:
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.hbase.HBaseSinkFunction;
import org.apache.flink.streaming.connectors.hbase.HBaseTableSource;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
public class FlinkHBaseExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
DataStream<String> stream = env.addSource(consumer);
stream.addSink(new SinkFunction<String>() {
@Override
public void invoke(String value, Context context) throws Exception {
Put put = new Put(Bytes.toBytes(key));
put.addColumn(family, qualifier, Bytes.toBytes(value));
hTable.put(put);
}
});
env.execute("Flink HBase Example");
}
}
Apache Kafka
Kafka 是一个分布式流处理平台。以下是使用Kafka的Java Producer发送消息的示例:
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public class KafkaProducerExample {
public static void main(String[] args) {
Producer<String, String> producer = new KafkaProducer<>(properties);
for (int i = 0; i < 100; i++)
producer.send(new ProducerRecord<>("topic", Integer.toString(i), "message " + i));
producer.close();
}
}
Apache HBase
HBase 是一个分布式的、版本化的、非关系的数据存储系统。以下是使用HBase的Java API进行写入的示例:
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
public class HBaseExample {
public static void main(String[] args) throws Exception {
评论已关闭