1、pom.xml
<properties>
<java.version>1.8</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spring-boot.version>2.3.7.RELEASE</spring-boot.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-hbase_2.12</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.11</artifactId>
<version>1.14.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.10.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web_2.11</artifactId>
<version>1.9.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
2、生产者JProducer
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Date;
import java.util.Properties;
/***
* *往Kafka写入数据
* *json串
* *主题topic:flink_topic
*/
public class JProducer extends Thread {
public static final String broker_list = "localhost:9094";
public static final String topic = "flink_topic"; //kafka topic 需要和 消费者程序用同一个 topic
public static void main(String[] args) {
JProducer jproducer = new JProducer();
jproducer.start();
}
@Override
public void run() {
producer();
}
private void producer() {//向kafka批量生成记录
Properties props = config();//kafka连接
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
String json = "{\"id\":" + i + ",\"ip\":\"192.168.0." + i + "\",\"date\":" + new Date().toString() + "}";
String k = "key" + i;
producer.send(new ProducerRecord<String, String>(topic, k, json));
}
producer.close();
}
private Properties config() { //kafka连接
Properties props = new Properties();
props.put("bootstrap.servers",broker_list);
props.put("acks", "1");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
return props;
}
}
3、消费者FlinkHBase
//读取Kafka数据，主题topic:flink_topic
//存入hbase
import org.apache.commons.net.ntp.TimeStamp;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
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.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.Date;
import java.util.Properties;
public class FlinkHBase {
private static String zkServer = "127.0.0.1";
private static String zkPort = "2181";
private static TableName tableName = TableName.valueOf("testflink");//hbase 表格
private static final String cf = "ke";//hbase 表格区
private static final String topic = "flink_topic"; //kafka主题
public static final String broker_list = "localhost:9094";
public static void main(String[] args) {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(1000);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//读取Kafka数据，主题topic:flink_topic
DataStream<String> transction = env.addSource(new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), props()).setStartFromEarliest());
transction.rebalance().map(new MapFunction<String, Object>() {
private static final long serialVersionUID = 1L;
@Override
public String map(String value) throws IOException {
write2HBase(value);//读取Kafka数据 存入hbase
return value;
}
}).print();
try {
env.execute();
} catch (Exception ex) {
ex.printStackTrace();
}
}
public static Properties props() {
Properties props = new Properties();
props.put("bootstrap.servers", broker_list);
props.put("zookeeper.connect", "localhost:2182");
props.put("group.id", "kv_flink");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
return props;
}
//读取Kafka数据 存入hbase
public static void write2HBase(String value) throws IOException {
Configuration config = HBaseConfiguration.create();
Connection conn=null;
config.set("hbase.zookeeper.quorum", zkServer);
config.set("hbase.zookeeper.property.clientPort", zkPort);
//config.set("hbase.master","127.0.0.1：16010");
/*config.setInt("hbase.rpc.timeout", 30000);
config.setInt("hbase.client.operation.timeout", 30000);
config.setInt("hbase.client.scanner.timeout.period", 30000);*/
try {
conn= ConnectionFactory.createConnection(config);
}catch (IOException e){
e.printStackTrace();;
}
HBaseAdmin admin =new HBaseAdmin(config);
if (!admin.tableExists(tableName)) {//表testflink是否存在
admin.createTable(new HTableDescriptor(tableName).addFamily(new HColumnDescriptor(cf)));
}
Table table = conn.getTable(tableName);//testflink hbase 表格
TimeStamp ts = new TimeStamp(new Date());
Date date = ts.getDate();
Put put2 = new Put(Bytes.toBytes(date.getTime()));
put2.addColumn(Bytes.toBytes(cf), Bytes.toBytes("test"), Bytes.toBytes(value));
table.put(put2);
table.close();
conn.close();
}
}
