苏州住房与城乡建设局网站,桂林旅游景点,制作网页焦点图,哪里有做兼职的网站3.7.基于Flink将数据写入到HBase
3.7.1.编写Flink完成数据写入到Hbase操作, 完成数据备份, 便于后续进行即席查询和离线分析
3.7.1.1.HBase基本介绍
hbase是基于Google发布bigTable论文产生一款软件, 是一款noSQL型数据, 不支持SQL. 不支持join的操作, 没有表关系, 不支持事…3.7.基于Flink将数据写入到HBase
3.7.1.编写Flink完成数据写入到Hbase操作, 完成数据备份, 便于后续进行即席查询和离线分析
3.7.1.1.HBase基本介绍
hbase是基于Google发布bigTable论文产生一款软件, 是一款noSQL型数据, 不支持SQL. 不支持join的操作, 没有表关系, 不支持事务(多行事务),hbase是基于 HDFS的采用java 语言编写
查询hbase数据一般有三种方案(主键(row key)查询, 主键的范围检索,查询全部数据)
都是以字节类型存储存储结构化和半结构化数据。
hbase表的特点: 大 面向列的存储方案 稀疏性
2.7.1.2.应用场景
1需要进行随机读写的操作。 2数据量比较大。 3数据比较稀疏。
2.7.1.3.HBase安装操作
本次安装的HBase为2.2.7,详细的安装手册大家可以参考资料, 还需要大家注意,HBase的启动需要依赖于zookeeper 和HDFS的, 顾需要先安装 HADOOP与zookeeper
1-在Hbase中创建目标表
create itcast_h_ems, {NAMEf1,COMPRESSIONGZ},{NUMREGIONS6, SPLITALGOHexStringSplit}2- 编写Flink代码完成写入Hbase操作
import com.itheima.pojo.PulsarTopicPojo;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.pulsar.FlinkPulsarSource;
import org.apache.flink.streaming.connectors.pulsar.internal.JsonDeser;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import java.util.Properties;// 基于Flink消费Pulsar数据, 然后将数据灌入到HBase中, 完成数据备份, 以及后续即席查询和离线分析
public class ItcastFlinkToHBase {public static void main(String[] args) throws Exception {//1. 创建Flinnk流式处理核心环境类对象 和 Table API 核心环境类对象StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tableEnv StreamTableEnvironment.create(env);//2. 添加Source组件, 从Pulsar中读取消息数据Properties props new Properties();props.setProperty(topic,persistent://public/default/itcast_ems_tab);props.setProperty(partition.discovery.interval-millis,5000);FlinkPulsarSourcePulsarTopicPojo pulsarSource new FlinkPulsarSourcePulsarTopicPojo(pulsar://node1:6650,node2:6650,node3:6650,http://node1:8080,node2:8080,node3:8080,JsonDeser.of(PulsarTopicPojo.class),props);//2.1 设置pulsarSource组件在消费数据的时候, 默认从什么位置开始消费pulsarSource.setStartFromLatest();DataStreamSourcePulsarTopicPojo dataStreamSource env.addSource(pulsarSource);//2.2 转换为Flink TableSchema schema Schema.newBuilder().column(id, DataTypes.INT()).column(sid, DataTypes.STRING()).column(ip, DataTypes.STRING()).column(session_id, DataTypes.STRING()).column(create_time, DataTypes.STRING()).column(yearInfo, DataTypes.STRING()).column(monthInfo, DataTypes.STRING()).column(dayInfo, DataTypes.STRING()).column(hourInfo, DataTypes.STRING()).column(seo_source, DataTypes.STRING()).column(area, DataTypes.STRING()).column(origin_channel, DataTypes.STRING()).column(msg_count, DataTypes.INT()).column(from_url, DataTypes.STRING()).build();tableEnv.createTemporaryView(itcast_ems,dataStreamSource,schema);//2.3: 定义HBase的目标表String hTable create table itcast_h_ems(rowkey int,f1 ROWsid STRING,ip STRING,session_id STRING,create_time STRING,yearInfo STRING,monthInfo STRING,dayInfo STRING,hourInfo STRING,seo_source STRING,area STRING,origin_channel STRING,msg_count INT,from_url STRING,primary key(rowkey) NOT ENFORCED ) WITH (connectorhbase-2.2,table-nameitcast_h_ems,zookeeper.quorumnode1:2181,node2:2181,node3:2181);//4. 执行操作tableEnv.executeSql(hTable);tableEnv.executeSql(insert into itcast_h_ems select id,ROW(sid,ip,session_id,create_time,yearInfo,monthInfo,dayInfo,hourInfo,seo_source,area,origin_channel,msg_count,from_url) from itcast_ems);}}PulsarTopicPojo
public class PulsarTopicPojo {private Integer id;private String sid;private String ip;private String session_id;private String create_time;private String yearInfo;private String monthInfo;private String dayInfo;private String hourInfo;private String seo_source;private String area;private String origin_channel;private Integer msg_count;private String from_url;public PulsarTopicPojo() {}public PulsarTopicPojo(Integer id, String sid, String ip, String session_id, String create_time, String yearInfo, String monthInfo, String dayInfo, String hourInfo, String seo_source, String area, String origin_channel, Integer msg_count, String from_url) {this.id id;this.sid sid;this.ip ip;this.session_id session_id;this.create_time create_time;this.yearInfo yearInfo;this.monthInfo monthInfo;this.dayInfo dayInfo;this.hourInfo hourInfo;this.seo_source seo_source;this.area area;this.origin_channel origin_channel;this.msg_count msg_count;this.from_url from_url;}public void setData(Integer id, String sid, String ip, String session_id, String create_time, String yearInfo, String monthInfo, String dayInfo, String hourInfo, String seo_source, String area, String origin_channel, Integer msg_count, String from_url) {this.id id;this.sid sid;this.ip ip;this.session_id session_id;this.create_time create_time;this.yearInfo yearInfo;this.monthInfo monthInfo;this.dayInfo dayInfo;this.hourInfo hourInfo;this.seo_source seo_source;this.area area;this.origin_channel origin_channel;this.msg_count msg_count;this.from_url from_url;}public Integer getId() {return id;}public void setId(Integer id) {this.id id;}public String getSid() {return sid;}public void setSid(String sid) {this.sid sid;}public String getIp() {return ip;}public void setIp(String ip) {this.ip ip;}public String getSession_id() {return session_id;}public void setSession_id(String session_id) {this.session_id session_id;}public String getCreate_time() {return create_time;}public void setCreate_time(String create_time) {this.create_time create_time;}public String getYearInfo() {return yearInfo;}public void setYearInfo(String yearInfo) {this.yearInfo yearInfo;}public String getMonthInfo() {return monthInfo;}public void setMonthInfo(String monthInfo) {this.monthInfo monthInfo;}public String getDayInfo() {return dayInfo;}public void setDayInfo(String dayInfo) {this.dayInfo dayInfo;}public String getHourInfo() {return hourInfo;}public void setHourInfo(String hourInfo) {this.hourInfo hourInfo;}public String getSeo_source() {return seo_source;}public void setSeo_source(String seo_source) {this.seo_source seo_source;}public String getArea() {return area;}public void setArea(String area) {this.area area;}public String getOrigin_channel() {return origin_channel;}public void setOrigin_channel(String origin_channel) {this.origin_channel origin_channel;}public Integer getMsg_count() {return msg_count;}public void setMsg_count(Integer msg_count) {this.msg_count msg_count;}public String getFrom_url() {return from_url;}public void setFrom_url(String from_url) {this.from_url from_url;}Overridepublic String toString() {return PulsarTopicPojo{ id id , sid sid \ , ip ip \ , session_id session_id \ , create_time create_time \ , yearInfo yearInfo \ , monthInfo monthInfo \ , dayInfo dayInfo \ , hourInfo hourInfo \ , seo_source seo_source \ , area area \ , origin_channel origin_channel \ , msg_count msg_count , from_url from_url \ };}
}
