flink连接kafka整合hbase，scala

解析kafka当中的json格式的数据，入hbase

import java.util.Propertiesimport com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011object IncrementOrder {def main(args: Array[String]): Unit = {val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//隐式转换import org.apache.flink.api.scala._//checkpoint配置environment.enableCheckpointing(100);environment.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);environment.getCheckpointConfig.setMinPauseBetweenCheckpoints(500);environment.getCheckpointConfig.setCheckpointTimeout(60000);environment.getCheckpointConfig.setMaxConcurrentCheckpoints(1);environment.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);environment.setStateBackend(new RocksDBStateBackend("hdfs://node01:8020/flink_kafka/checkpoints",true));val props = new Propertiesprops.put("bootstrap.servers", "node01:9092")props.put("zookeeper.connect", "node01:2181")props.put("group.id", "flinkHouseGroup")props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")props.put("auto.offset.reset", "latest")props.put("flink.partition-discovery.interval-millis", "30000")val kafkaSource = new FlinkKafkaConsumer011[String]("flink_house",new SimpleStringSchema(),props)kafkaSource.setCommitOffsetsOnCheckpoints(true)//设置statebackendval result: DataStream[String] = environment.addSource(kafkaSource)val orderResult: DataStream[OrderObj] = result.map(x => {val jsonObj: JSONObject = JSON.parseObject(x)val database: AnyRef = jsonObj.get("database")val table: AnyRef = jsonObj.get("table")val `type`: AnyRef = jsonObj.get("type")val string: String = jsonObj.get("data").toStringOrderObj(database.toString,table.toString,`type`.toString,string)})orderResult.addSink(new HBaseSinkFunction)environment.execute()}
}
case class OrderObj(database:String,table:String,`type`:String,data:String) extends Serializable

定义插入数据到hbase的程序

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.conf
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client._class HBaseSinkFunction extends RichSinkFunction[OrderObj]{var connection:Connection = _var hbTable:Table  =  _override def open(parameters: Configuration): Unit = {val configuration: conf.Configuration = HBaseConfiguration.create()configuration.set("hbase.zookeeper.quorum", "node01,node02,node03")configuration.set("hbase.zookeeper.property.clientPort", "2181")connection = ConnectionFactory.createConnection(configuration)hbTable = connection.getTable(TableName.valueOf("flink:data_orders"))}override def close(): Unit = {if(null != hbTable){hbTable.close()}if(null != connection){connection.close()}}def insertHBase(hbTable: Table, orderObj: OrderObj) = {val database: String = orderObj.databaseval table: String = orderObj.tableval value: String = orderObj.`type`val orderJson: JSONObject = JSON.parseObject(orderObj.data)val orderId: String = orderJson.get("orderId").toStringval orderNo: String = orderJson.get("orderNo").toStringval userId: String = orderJson.get("userId").toStringval goodId: String = orderJson.get("goodId").toStringval goodsMoney: String = orderJson.get("goodsMoney").toStringval realTotalMoney: String = orderJson.get("realTotalMoney").toStringval payFrom: String = orderJson.get("payFrom").toStringval province: String = orderJson.get("province").toStringval createTime: String = orderJson.get("createTime").toStringval put = new Put(orderId.getBytes())put.addColumn("f1".getBytes(),"orderNo".getBytes(),orderNo.getBytes())put.addColumn("f1".getBytes(),"userId".getBytes(),userId.getBytes())put.addColumn("f1".getBytes(),"goodId".getBytes(),goodId.getBytes())put.addColumn("f1".getBytes(),"goodsMoney".getBytes(),goodsMoney.getBytes())put.addColumn("f1".getBytes(),"realTotalMoney".getBytes(),realTotalMoney.getBytes())put.addColumn("f1".getBytes(),"payFrom".getBytes(),payFrom.getBytes())put.addColumn("f1".getBytes(),"province".getBytes(),province.getBytes())put.addColumn("f1".getBytes(),"createTime".getBytes(),createTime.getBytes())
/*
*
* */hbTable.put(put);}def deleteHBaseData(hbTable: Table, orderObj: OrderObj) = {val orderJson: JSONObject = JSON.parseObject(orderObj.data)val orderId: String = orderJson.get("orderId").toStringval delete = new Delete(orderId.getBytes())hbTable.delete(delete)}override def invoke(orderObj: OrderObj, context: SinkFunction.Context[_]): Unit = {val database: String = orderObj.databaseval table: String = orderObj.tableval typeResult: String = orderObj.`type`if(database.equalsIgnoreCase("product")  &&  table.equalsIgnoreCase("kaikeba_orders")){if(typeResult.equalsIgnoreCase("insert")){//插入hbase数据insertHBase(hbTable,orderObj)}else if(typeResult.equalsIgnoreCase("update")){//更新hbase数据insertHBase(hbTable,orderObj)}else if(typeResult.equalsIgnoreCase("delete")){//删除hbase数据deleteHBaseData(hbTable,orderObj)}}}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/508769.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

sparkStreaming连接kafka整合hbase和redis

sparkStreaming连接kafka整合hbase和redis

sparkStreaming消费kafka数据，并将数据保存到redis和hbase当中去，实现实时 import org.apache.hadoop.hbase.client.{Admin, Connection} import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName} import org.apache.kafka.c…
阅读更多...
sparksql一些指标

sparksql一些指标

统计指标 select substr(tb.begin_address_code , 1 ,4) as begin_address_code , count(distinct vehicle_license) as dayVehicleCount from (select begin_address_code , vehicle_license from order where date_format(create_time , yyyy-MM-dd) 2020-02-15 ) tb grou…
阅读更多...
sparkConf常见参数设置

sparkConf常见参数设置

def getSparkConf():SparkConf {val sparkConf: SparkConf new SparkConf().set("spark.driver.cores","4") //设置driver的CPU核数.set("spark.driver.maxResultSize","2g") //设置driver端结果存放的最大容量，这里设置…
阅读更多...
sparkSession常见参数设置

sparkSession常见参数设置

def getSparkSession(sparkConf:SparkConf):SparkSession {val sparkSession: SparkSession SparkSession.builder().config(sparkConf)//调度模式.config("spark.scheduler.mode", "FAIR").config("spark.executor.memoryOverhead", "51…
阅读更多...
关于kafka中acks是否可以为all

关于kafka中acks是否可以为all

kafka源码中有这样一段代码： org.apache.kafka.clients.producer.KafkaProducer private static int parseAcks(String acksString) {try {return acksString.trim().equalsIgnoreCase("all") ? -1 : Integer.parseInt(acksString.trim());} catch (Numb…
阅读更多...
关于统计时间切片标签的一些sql

关于统计时间切片标签的一些sql

------当天付费明细表 DROP TABLE IF EXISTS rpt.tmp_mm_rb_daily_ffmx; create table rpt.tmp_mm_rb_daily_ffmx as select a.* FROM (select c.feemsisdn, c.destmsisdn, c.day, c.price/1000 fee, c.contentid, dc.content_name, c.ordernumber, c.cdrtime, c.createtime, c…
阅读更多...
hadoop 二次开发DatanodeWriteTimeout设置

hadoop 二次开发DatanodeWriteTimeout设置

int getDatanodeWriteTimeout(int numNodes) {return this.dfsClientConf.confTime > 0 ? this.dfsClientConf.confTime 5000 * numNodes : 0;}int getDatanodeReadTimeout(int numNodes) {return this.dfsClientConf.socketTimeout > 0 ? 5000 * numNodes this.dfsC…
阅读更多...
聚类算法

聚类算法

假定样本集 D {X1， 的，…， Xm} 包含 m 个无标记样本， 每个样本 X (X1; X2;… ; Xn) 是一个 n 维特征向量，则聚类算法将样本 集 D 划分为 k 个不相交的簇 {Gl I l 1， 2;… ，时，其中…
阅读更多...
k-means均值向量

k-means均值向量

给定样本集 D {Xl) 的，… ,xm}, “k 均值” (k-means )算法针对聚类所 得簇划分 C {C1, C2,…, Ck} 最小化平方误差 ι ELL Ilx 一队IIL il EGi 其中队甘il LEGi X 是簇 q 的均值向量.在一定程度上 刻画了簇内样本围绕簇均值向量的紧密程度， E 值越小则…
阅读更多...
学习向量量化

学习向量量化

与 k 均值算法类似，“学习向量量化” (Learning Vector Quantization，简 称 LVQ)也是试图找到一组原型向量来刻画聚类结构， 但与一般聚类算法不同 的是， LVQ 假设数据样本带有类别标记，学习过程利用样本的这些监督信息…
阅读更多...
k 近邻加权平均

k 近邻加权平均

k 近邻(k-Nearest Neighbor，简称 kNN)学习是一种常用的监督学习方法， 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本，然后基于这 k 个"邻居"的信息来进行预测. 通常， 在分…
阅读更多...
k 近邻降维

k 近邻降维

k 近邻(k-Nearest Neighbor，简称 kNN)学习是一种常用的监督学习方法， 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本，然后基于这 k 个"邻居"的信息来进行预测. 通常， 在分…
阅读更多...
维度建模工具

维度建模工具

幵始维度建模工作前，项目组需要理解业务需求，以及作为基础的源数据的实际情况。 通过与、 Ik务代表交流来发现需求，用于理解他们的基于关键性能指标、竞争性商业问题、 决策制定过程、支持分析需求的目标。同时，数据实际情况可以通…
阅读更多...
Cube和Grouping 和Rollup

Cube和Grouping 和Rollup

增强的聚合 Cube和Grouping 和Rollup 这几个分析函数通常用于OLAP中，不能累加，而且需要根据不同维度上钻和下钻的指标统计，比如，分小时、天、月的UV数。 GROUPING SETS 在一个GROUP BY查询中，根据不同的维度组合进行聚…
阅读更多...
常见维度建模错误

常见维度建模错误

需要避免的常见维度建模错误 错误 10: 在事实表中放入文本属性 要从数据仓库事实表中 挑出这些文本属性，并将它们放入维度表中。 错误 9: 限制使用冗长的描述符以节省空间 维度表从几何上看总是比事实表小很多。 错误 8: 将层次划分为多个维度 以用户看来最自然最 有…
阅读更多...
2020-09-21

2020-09-21

columns has 234 elements while hbase.columns.mapping has 92 elements (counting the key if implicit)) 根本原因： 对于4000个字符，hive Metastore中SERDE_PARAMS表中PARAM_VALUE字段的字符限制是此问题的根本原因。 此限制可防止Hive创建具有高列数…
阅读更多...
2020-09-23

2020-09-23

insert into table ads_user_action_convert_day select ‘2019-02-10’, uv.day_count, ua.order_count, cast(ua.order_count/uv.day_count as decimal(10,2)) visitor2order_convert_ratio, ua.payment_count, cast(ua.payment_count/ua.order_count as decimal(10,2)) orde…
阅读更多...
2020-09-28

2020-09-28

Mybatis中#{}和${} 1、#{}将传入的数据都当成一个字符串，会对自动传入的数据加一个引号（单引号？双引号？加了引号就对了）如： //传入的值是sex order by #{column} 解析后为 order by “sex” //将会出错 2…
阅读更多...
PageUtil

PageUtil

Data NoArgsConstructor public class PageUtil implements Serializable {private static final long serialVersionUID 1L;/*** 总记录数*/private int totalCount;/*** 每页记录数*/private int pageSize;/*** 总页数*/private int totalPage;/*** 当前页数*/private int c…
阅读更多...
springboot+事务，多张表的操作事务回滚

springboot+事务，多张表的操作事务回滚

第一步，在springboot的启动类上开启事务，注解 EnableTransactionManagement 第二步：事务注解，回滚 Transactional(rollbackFor Exception.class) //Transactional和try catch捕获异常会让注解失效，所以： 第…
阅读更多...
最新文章

Copyright @ 2022~2023