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)}}}
}

flink连接kafka整合hbase，scala相关推荐

sparkStreaming连接kafka整合hbase和redis
sparkStreaming消费kafka数据,并将数据保存到redis和hbase当中去,实现实时 import org.apache.hadoop.hbase.client.{Admin, Con ...
flink连接kafka报错
问题描述: 通过本地idea方式连接kafka,报如下错误: 可能原因:本地程序会查询本地配置,看是否存在ip映射,可能是本地ip映射不存在,导致连接kafka失败.(主要是由于kafka的serve ...
flink入门3-Flink连接Kafka、Redis，实现Kafka Source/Redis Sink
本篇文章将会一步步实现如何使用Flink对接Kafka和Redis,并将Kafka中的数据存储到Redis中,这种场景也是真实项目会遇到的. 1.单机部署Kafka 1.1 下载Kafka压缩包,解压 ...
Flink系列之：基于scala语言实现flink实时消费Kafka Topic中的数据
Flink系列之:基于scala语言实现flink实时消费Kafka Topic中的数据一.引入flink相关依赖二.properties保存连接kafka的配置三.构建flink实时消费环境 ...
2.4-2.5、Hive整合（整合Spark、整合Hbase）、连接方式Cli、HiveServer和hivemetastore、Squirrel SQL Client等
2.4其它整合 2.4.1Hive整合Spark Spark整合hive,需要将hive_home下的conf下的hive_site.xml放到spark_home下的conf目录下.(3台服务器都做 ...
大数据开发超高频面试题！大厂面试必看！包含Hadoop、zookeeper、Hive、flume、kafka、Hbase、flink、spark、数仓等
大数据开发面试题包含Hadoop.zookeeper.Hive.flume.kafka.Hbase.flink.spark.数仓等高频面试题. 数据来自原博主爬虫获取! 文章目录大数据开发面试题 ...
Flink 系例之 Connectors 连接 Kafka
通过使用 Flink DataStream Connectors 数据流连接器连接到 ElasticSearch 搜索引擎的文档数据库 Index,并提供数据流输入与输出操作: 示例环境 java.v ...
kafka学习_《从0到1学习Flink》—— Flink 读取 Kafka 数据写入到 RabbitMQ
前言之前有文章 <从0到1学习Flink>-- Flink 写入数据到 Kafka 写过 Flink 将处理后的数据后发到 Kafka 消息队列中去,当然我们常用的消息队列可不止这一种, ...
flink checkpoint 恢复_Flink解析 | Apache Flink结合Kafka构建端到端的ExactlyOnce处理
周凯波(宝牛) 阿里巴巴技术专家,四川大学硕士,2010年毕业后加入阿里搜索事业部,从事搜索离线平台的研发工作,参与将搜索后台数据处理架构从MapReduce到Flink的重构.目前在阿里计算平台事业 ...

flink连接kafka整合hbase，scala

flink连接kafka整合hbase，scala相关推荐

最新文章

热门文章