15_基于Flink将pulsar数据写入到ClickHouse
3.8.基于Flink将数据写入到ClickHouse
编写Flink完成数据写入到ClickHouse操作, 后续基于CK完成指标统计操作
3.8.1.ClickHouse基本介绍
ClickHouse 是俄罗斯的Yandex于2016年开源的列式存储数据库(DBMS),使用C++语言编写,主要用于在线分析处理查询(OLAP),能够使用SQL查询实时生成分析数据报告。

结论: ClickHouse像很多OLAP数据库一样,单表查询速度由于关联查询,而且ClickHouse的两者差距更为明显。
3.8.2.ClickHouse安装步骤
本项目中,我们仅需要安装单机测试版本即可使用(node2安装), 在实际生产中, 大家可以直接将分布式集群版本
- 1-设置yum源
sudo yum install yum-utils
sudo rpm --import https://repo.clickhouse.com/CLICKHOUSE-KEY.GPG
sudo yum-config-manager --add-repo https://repo.clickhouse.com/rpm/stable/x86_64
- 2- 直接基于yum安装即可
sudo yum install clickhouse-server clickhouse-client
- 3-修改配置文件
vim /etc/clickhouse-server/config.xml
修改178行: 打开这一行的注释
<listen_host>::</listen_host>

- 4-启动clickhouse的server
systemctl start clickhouse-server
停止:
systemctl stop clickhouse-server
重启
systemctl restart clickhouse-server
- 5-进入客户端

3.8.3.在ClickHouse中创建目标表
create database itcast_ck;
use itcast_ck;
create table itcast_ck.itcast_ck_ems(
id int,
sid varchar(128),
ip varchar(128),
create_time varchar(128),
session_id varchar(128),
yearInfo varchar(128),
monthInfo varchar(128),
dayInfo varchar(128),
hourInfo varchar(128),
seo_source varchar(128),
area varchar(128),
origin_channel varchar(128),
msg_count int(128),
from_url varchar(128),
PRIMARY KEY (`id`)
) ENGINE=ReplacingMergeTree();
3.8.4.编写Flink代码完成写入到CK操作
import com.itheima.pojo.PulsarTopicPojo;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
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.types.Row;import java.sql.Types;
import java.util.Properties;// 基于Flink完成读取Pulsar中数据将消息数据写入到clickhouse中
public class ItcastFlinkToClickHouse {public static void main(String[] args) throws Exception {//1. 创建Flinnk流式处理核心环境类对象 和 Table API 核心环境类对象StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2. 添加Source组件, 从Pulsar中读取消息数据Properties props = new Properties();props.setProperty("topic","persistent://public/default/itcast_ems_tab");props.setProperty("partition.discovery.interval-millis","5000");FlinkPulsarSource<PulsarTopicPojo> pulsarSource = new FlinkPulsarSource<PulsarTopicPojo>("pulsar://node1:6650,node2:6650,node3:6650","http://node1:8080,node2:8080,node3:8080",JsonDeser.of(PulsarTopicPojo.class),props);//2.1 设置pulsarSource组件在消费数据的时候, 默认从什么位置开始消费pulsarSource.setStartFromLatest();DataStreamSource<PulsarTopicPojo> dataStreamSource = env.addSource(pulsarSource);//2.2 转换数据操作: 将 PulsarTopicPojo 转换为ROW对象SingleOutputStreamOperator<Row> rowDataSteam = dataStreamSource.map(new MapFunction<PulsarTopicPojo, Row>() {@Overridepublic Row map(PulsarTopicPojo pulsarTopicPojo) throws Exception {return Row.of(pulsarTopicPojo.getId(), pulsarTopicPojo.getSid(), pulsarTopicPojo.getIp(), pulsarTopicPojo.getCreate_time(),pulsarTopicPojo.getSession_id(), pulsarTopicPojo.getYearInfo(), pulsarTopicPojo.getMonthInfo(), pulsarTopicPojo.getDayInfo(),pulsarTopicPojo.getHourInfo(), pulsarTopicPojo.getSeo_source(), pulsarTopicPojo.getArea(), pulsarTopicPojo.getOrigin_channel(),pulsarTopicPojo.getMsg_count(), pulsarTopicPojo.getFrom_url());}});//2.3: 设置sink操作写入到CK操作String insertSql = "insert into itcast_ck.itcast_ck_ems (id,sid,ip,create_time,session_id,yearInfo,monthInfo,dayInfo,hourInfo,seo_source,area,origin_channel,msg_count,from_url) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)";JDBCAppendTableSink tableSink = JDBCAppendTableSink.builder().setDrivername("ru.yandex.clickhouse.ClickHouseDriver").setDBUrl("jdbc:clickhouse://node2:8123/itcast_ck").setQuery(insertSql).setBatchSize(1).setParameterTypes(Types.INTEGER,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.VARCHAR,Types.INTEGER,Types.VARCHAR).build();tableSink.emitDataStream(rowDataSteam);//3. 提交执行env.execute("itcast_to_ck");}
}
3.9.HBase对接Phoenix实现即席查询
3.9.1.Phoenix安装操作
Phoenix是属于apache旗下的一款基于hbase的工具, 此工具提供一种全新的方式来操作hbase中数据(SQL),
同时Phoenix对hbase进行大量的优化工作, 能够让我们更加有效的操作hbase
整个安装操作, 大家可以参考资料中安装手册, 进行安装即可
3.9.2.在Phoenix中创建表
create view "itcast_h_ems" (
"id" integer primary key,
"f1"."sid" varchar,
"f1"."ip" varchar,
"f1"."create_time" varchar,
"f1"."session_id" varchar,
"f1"."yearInfo" varchar,
"f1"."monthInfo" varchar,
"f1"."dayInfo" varchar,
"f1"."hourInfo" varchar,
"f1"."seo_source" varchar,
"f1"."area" varchar,
"f1"."origin_channel" varchar,
"f1"."msg_count" integer,
"f1"."from_url" varchar
);
3.9.3.在Phoenix中类型说明

相关文章:
15_基于Flink将pulsar数据写入到ClickHouse
3.8.基于Flink将数据写入到ClickHouse 编写Flink完成数据写入到ClickHouse操作, 后续基于CK完成指标统计操作 3.8.1.ClickHouse基本介绍 ClickHouse 是俄罗斯的Yandex于2016年开源的列式存储数据库(DBMS),使用C语言编写,主要用…...
Pycharm如何打断点进行调试?
断点调试,是编写程序中一个很重要的步骤,有些简单的程序使用print语句就可看出问题,而比较复杂的程序,函数和变量较多的情况下,这时候就需要打断点了,更容易定位问题。 一、添加断点 在代码的行标前面&…...
微服务02-docker
1、Docker架构 1.1 镜像和容器 Docker中有几个重要的概念: 镜像(Image):Docker将应用程序及其所需的依赖、函数库、环境、配置等文件打包在一起,称为镜像。Docker镜像是用于创建 Docker 容器的模板 。就像面向对象编…...
CSS:盒子模型 与 多种横向布局方法
目录 盒子模型块级盒子内联级盒子内联块级盒子弹性盒子display 改变模型区域划分text 内容区padding 填充区border 边框区margin 外边距直接设置盒子大小 布局横向布局方法一 float 浮起来方法二 内联块级元素实现方法三 弹性盒子模型 盒子模型 块级盒子 独占一行,…...
用node.js搭建一个视频推流服务
由于业务中有不少视频使用的场景,今天来说说如何使用node完成一个视频推流服务。 先看看效果: 这里的播放的视频是一个多个Partial Content组合起来的,每个Partial Content大小是1M。 一,项目搭建 (1)初…...
【SpringCloud】Feign远程调用
先来看我们以前利用RestTemplate发起远程调用的代码: String url "http://userservice/user/" order.getUserId(); User user restTemplate.getForObject(url, User.class);存在下面的问题: • 代码可读性差,编程体验不统一 • …...
集合Collection-List-ArrayList学习
一、集合 集合是数据容器。相较于数组集合具有以下几个特点: 数组一旦创建,长度不可改变。集合的长度会自动扩容。集合具有很多数组没有的功能函数API数组元素的存储特点单一,不同的集合有不同的存储特点。 1. Collection顶层接口 Collect…...
mybatispuls代码生成器
引入依赖 <?xml version"1.0" encoding"UTF-8"?> <project xmlns"http://maven.apache.org/POM/4.0.0" xmlns:xsi"http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation"http://maven.apache.org/POM/4.0.…...
【设计模式】-代理模式
在软件开发中,经常遇到需要对某个对象进行控制或者监控的场景。而直接修改对象的代码可能使代码变得复杂且难以维护。这时,使用代理模式(Proxy Pattern)可以很好地解决这个问题。 代理模式是一种结构型设计模式,通过引…...
爬虫ip池越大越好吗?
作为一名资深的程序员,今天我要给大家分享一些关于爬虫ip池的知识。关于ip代理池的问题,答案是肯定的,池子越大越好。下面跟我一起来盘点一下ip池大的好处吧! 1、提高稳定性 爬虫ip池越大,意味着拥有更多可用的爬虫ip…...
目标检测常用的数据集格式
在目标检测领域,有三种常用的数据集: 数据集标注文件格式bbox格式vocxmlxmin, ymin, xmax, ymax:bbox左上角(xmin, ymin)和右下角(xmax, ymax)的坐标cocojsonx, y, w, h:bbox左上角坐标(x, y)以及宽(w)和高(h)yolotxtxcenter, ycenter, w, h:bbox的中心…...
chrome插件开发实例03-使用 chrome.storage API永久保存数据
目录 防止数据丢失 使用chrome.storage API 功能 功能演示 源代码 manifest.json popup.html...
Segment Anything(SAM) 计算过程
给定输入图像 I ∈ R 3 H W I \in R^{3 \times H \times W} I∈R3HW。给定需要的prompts: M ∈ R 1 H W M \in R^{1 \times H \times W} M∈R1HW,代表图片的前背景信息。 P ∈ R N 2 P \in R^{N \times 2} P∈RN2,其中 N N N 是点的个数…...
Nacos配置文件读取源码解析
Nacos配置文件读取 本篇文章是探究,springboot启动时nacos是如何将配置中心的配置读取到springboot环境中的 PropertySourceLocator org.springframework.cloud.bootstrap.config.PropertySourceLocator 是 springcloud 定义的一个顶级接口,用来定义所…...
Linux0.11内核源码解析-fcntl.c/iotcl.c/stat.c
fcntl fcntl.c实现了文件控制系统调用fcntl和两个文件句柄描述符的复制系统调用dup()和dup2()。 dup返回当前值最小的未用句柄,dup2返回指定新句柄的数值,句柄的复制操作主要用在文件的标准输入、输出重定向和管道方面。 dupfd 复制文件句柄ÿ…...
OpenStack简介
OpenStack简介 目录 OpenStack简介 1、云计算模式2、云计算 虚拟化 openstack之间的关系?3、OpenStack 中有哪些组件?4、计算节点负责虚拟机运行5、网络节点负责对外网络与内网之间的通信 5.1 网络节点仅包含Neutron服务5.2 网络节点包含三个网络端口6、…...
二分法的应用
文章目录 什么是二分法🎮二分查找的优先级二分查找的步骤💥图解演示🧩 代码演示🫕python程序实现🐈⬛C程序实现🐕🦺C程序实现🐯Java程序实现🐳 非常规类二分查找&…...
ChatGPT在大规模数据处理和信息管理中的应用如何?
ChatGPT作为一种强大的自然语言处理模型,在大规模数据处理和信息管理领域有着广泛的应用潜力。它可以利用其文本生成、文本理解和问答等能力,为数据分析、信息提取、知识管理等任务提供智能化的解决方案。以下将详细介绍ChatGPT在大规模数据处理和信息管…...
【算法篇C++实现】五大常规算法
文章目录 🚀一、分治法⛳(一)算法思想⛳(二)相关代码 🚀二、动态规划算法⛳(一)算法思想⛳(二)相关代码 🚀三、回溯算法⛳(一…...
MySQL和钉钉单据接口对接
MySQL和钉钉单据接口对接 数据源系统:钉钉 钉钉(DingTalk)是阿里巴巴集团打造的企业级智能移动办公平台,是数字经济时代的企业组织协同办公和应用开发平台。钉钉将IM即时沟通、钉钉文档、钉闪会、钉盘、Teambition、OA审批、智能人事、钉工牌…...
忍者像素绘卷保姆级教程:从Docker Compose启动到UI界面汉化配置
忍者像素绘卷保姆级教程:从Docker Compose启动到UI界面汉化配置 1. 环境准备与快速部署 在开始使用忍者像素绘卷之前,我们需要先准备好运行环境并完成部署。这个步骤非常简单,即使你是Docker新手也能轻松完成。 1.1 系统要求 确保你的系统…...
RV1126边缘设备性能实测:YOLOv8s vs YOLOv8m,谁才是性价比之王?
RV1126边缘设备性能实测:YOLOv8s与YOLOv8m的终极对决 在嵌入式AI领域,选择适合硬件平台的模型版本往往比模型本身更重要。当我们将目光投向Rockchip RV1126这类边缘计算设备时,YOLOv8系列中的s(small)和m(m…...
ZTP(零接触配置):实现自动化与高效的网络部署
在云计算、大数据和5G时代,网络基础设施的规模和复杂性大幅提升。传统的手动配置方式要求网络管理员逐台设备现场操作,效率低下、易出错且成本高昂。为应对这一挑战,ZTP(零接触配置)成为关键的自动化技术。ZTP允许新设…...
nimble 蓝牙开发二:BLE 协议栈核心组件 GAP/ATT/GATT 深度解析
1. BLE协议栈与Nimble架构全景透视 低功耗蓝牙(BLE)协议栈就像一座精心设计的建筑,每一层都有明确的职责分工。Nimble作为开源协议栈实现,其架构遵循蓝牙核心规范的同时,在资源占用和灵活性上做了大量优化。实测发现&a…...
Rusted PackFile Manager:从零开始打造你的Total War模组
Rusted PackFile Manager:从零开始打造你的Total War模组 【免费下载链接】rpfm Rusted PackFile Manager (RPFM) is a... reimplementation in Rust and Qt5 of PackFile Manager (PFM), one of the best modding tools for Total War Games. 项目地址: https://…...
GLM-5.1 深度解析:它为什么不只是一个更强的聊天模型?
GLM-5.1 深度解析:它为什么不只是一个更强的聊天模型?很多人看一个新模型,第一反应还是:它聊天强不强?推理行不行?中文味够不够? 但如果把视角放到 2026 年的大模型应用现场,你会发现…...
JW Player终极教程:如何5分钟实现网页视频播放
JW Player终极教程:如何5分钟实现网页视频播放 【免费下载链接】jwplayer No Longer Maintained 项目地址: https://gitcode.com/gh_mirrors/jw/jwplayer JW Player是一款功能强大的网页视频播放解决方案,已在超过200万个网站上使用,每…...
终极免费内容解锁工具:简单三步绕过所有付费墙限制
终极免费内容解锁工具:简单三步绕过所有付费墙限制 在数字化信息时代,你是否经常遇到这样的情况:一篇深度分析文章正看到关键处,突然弹出付费订阅提示?一个技术教程刚进入核心步骤,却被付费墙完全阻挡&…...
Phi-4-mini-reasoning与YOLOv5协同实战:图像描述生成与逻辑推理
Phi-4-mini-reasoning与YOLOv5协同实战:图像描述生成与逻辑推理 1. 效果亮点预览 当视觉识别遇上逻辑推理,会碰撞出怎样的火花?我们最近尝试了一个有趣的实验:用YOLOv5识别图片中的物体,再将识别结果输入Phi-4-mini-…...
Kandinsky-5.0-I2V-Lite-5s从零部署:JDK1.8环境下的Java客户端开发
Kandinsky-5.0-I2V-Lite-5s从零部署:JDK1.8环境下的Java客户端开发 1. 引言 还在用JDK1.8开发企业级应用?想集成最新的Kandinsky-5.0-I2V-Lite-5s图像转视频模型却担心兼容性问题?这篇教程就是为你准备的。我们将从零开始,手把手…...
