本文经授权转载自 PostgreSQL 中文社区。
Kafka 是分布式消息订阅系统，有非常好的横向扩展性，可实时存储海量数据，是流数据处理中间件的事实标准。当通过 Kafka 和 greenplum 搭建流处理管道时，如何高速可靠的完成流数据加载，成为用户最关心的问题。从 5.10 开始，Greenplum 发布了新的工具 GPKafka，为 Greenplum 提供了流数据加载的能力。
GPkafka 工具：kafka —> Greenplum
一、安装准备
kafka 安装：版本为 kafka_2.11-2.1.0。
greenplum 安装：版本为 5.16
二、Kafka 数据导入 GreenPlum
启动 kafka
# 启动zookeeper
$ /opt/zookeeper-3.4.12/bin/zkServer.sh start
# 启动kafka
$/opt/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon ../config/server.properties
复制代码
2.创建 gpss 扩展
在将 Kafka 消息数据加载到 Greenplum 数据库之前，必须在将 Kafka 数据写入 Greenplum 表的每个数据库中注册 Greenplum-Kafka 集成格式化程序函数；示例在 lottu 数据库
[gpadmin@oracle166 ~]$ psql
psql (8.3.23)
Type "help" for help.
lottu=# CREATE EXTENSION gpss;
复制代码
3.创建示例表
kafka 的数据格式 json 形式；样式：
{ "time": 1550198435941, "type": "type_mobileinfo", "phone_imei": "861738033581011", "phone_imsi": "", "phone_mac": "00:27:1c:95:47:09", "appkey": "307A5C626E6C2F6472636E6E6A2F736460656473", "phone_udid": "8F137BFFB2289784A5EA2DCADCE519C2", "phone_udid2": "744DD04CE29652F4F1D2DFFC8D3204A9", "appUdid": "D21C76419E54B18DDBB94BF2E6990183", "phone_resolution": "1280*720", "phone_apn": "", "phone_model": "BF T26", "phone_firmware_version": "5.1", "phone_softversion": "3.19.0", "phone_softname": "com.esbook.reader", "sdk_version": "3.1.8", "cpid": "blp1375_13621_001", "currentnetworktype": "wifi", "phone_city": "", "os": "android", "install_path": "\/data\/app\/com.esbook.reader-1\/base.apk", "last_cpid": "", "package_name": "com.esbook.reader", "src_code": "WIFIMAC:00:27:1c:95:47:09"}
复制代码
我需要其中的 package_name，appkey ，time， phone_udid，os， idfa，phone_imei，cpid，last_cpid，phone_number 字段；所以我创建的表语句
CREATE TABLE tbl_novel_mobile_log (
package_name text,
appkey text,
ts bigint,
phone_udid text,
os character varying(20),
idfa character varying(64),
phone_imei character varying(20),
cpid text,
last_cpid text,
phone_number character varying(20)
) ;
复制代码
4.创建 gpkafka.yaml 配置文件
gpkafka_mobile_yaml文件内容：
DATABASE: lottu
USER: gpadmin
HOST: oracle166
PORT: 5432
KAFKA:
INPUT:
SOURCE:
BROKERS: kafkaip:9092
TOPIC: mobile_info
COLUMNS:
- NAME: jdata
TYPE: json
FORMAT: json
ERROR_LIMIT: 10
OUTPUT:
TABLE: tbl_novel_mobile_log
MAPPING:
- NAME: package_name
EXPRESSION: (jdata->>'package_name')::text
- NAME: appkey
EXPRESSION: (jdata->>'appkey')::text
- NAME: ts
EXPRESSION: (jdata->>'time')::bigint
- NAME: phone_udid
EXPRESSION: (jdata->>'phone_udid')::text
- NAME: os
EXPRESSION: (jdata->>'os')::text
- NAME: idfa
EXPRESSION: (jdata->>'idfa')::text
- NAME: phone_imei
EXPRESSION: (jdata->>'phone_imei')::text
- NAME: cpid
EXPRESSION: (jdata->>'cpid')::text
- NAME: last_cpid
EXPRESSION: (jdata->>'last_cpid')::text
- NAME: phone_number
EXPRESSION: (jdata->>'phone_number')::text
COMMIT:
MAX_ROW: 1000
复制代码
5.创建 mobile_info topic
/opt/kafka/kafka_2.11-2.1.0/bin/kafka-topics.sh --create --zookeeper kafkaIp:2181 --replication-factor 1 --partitions 1 --topic mobile_info
复制代码
6.创建 kafka 的发布者
执行下列命令；并添加 kafka 记录
[root@oracle166 ~]# /opt/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh --broker-list kafkaIP:9092 --topic mobile_info
复制代码
>{"time":1550198435941,"type":"type_mobileinfo","phone_imei":"861738033581011","phone_imsi":"","phone_mac":"00:27:1c:95:47:09","appkey":"307A5C626E6C2F6472636E6E6A2F736460656473","phone_udid":"8F137BFFB2289784A5EA2DCADCE519C2","phone_udid2":"744DD04CE29652F4F1D2DFFC8D3204A9","appUdid":"D21C76419E54B18DDBB94BF2E6990183","phone_resolution":"1280*720","phone_apn":"","phone_model":"BFT26","phone_firmware_version":"5.1","phone_softversion":"3.19.0","phone_softname":"com.esbook.reader","sdk_version":"3.1.8","cpid":"blp1375_13621_001","currentnetworktype":"wifi","phone_city":"","os":"android","install_path":"\/data\/app\/com.esbook.reader1\/base.apk","last_cpid":"","package_name":"com.esbook.reader","src_code":"WIFIMAC:00:27:1c:95:47:09"}
{"time":1550198437885,"type":"type_mobileinfo","phone_imei":"862245038046551","phone_imsi":"","phone_mac":"02:00:00:00:00:00","appkey":"307A5C626F2F76646B74606F2F736460656473","phone_udid":"A3BB70A0218AEFC7908B1D79C0C02D77","phone_udid2":"E3976E0453010FC7F32B6143AA3A164E","appUdid":"4FBEF77BC076254ED0407CAD653E6954","phone_resolution":"1920*1080","phone_apn":"","phone_model":"LeX620","phone_firmware_version":"6.0","phone_softversion":"1.9.0","phone_softname":"cn.wejuan.reader","sdk_version":"3.1.8","cpid":"blf1298_14411_001","currentnetworktype":"wifi","phone_city":"","os":"android","install_path":"\/data\/app\/cn.wejuan.reader1\/base.apk","last_cpid":"","package_name":"cn.wejuan.reader","src_code":"ffffffff-9063-8e34-0000-00007efffeff"}
{"time":1550198438311,"type":"type_mobileinfo","phone_number":"","phone_imei":"867520045576831","phone_imsi":"460001122544742","phone_mac":"02:00:00:00:00:00","appkey":"307A5C626E6C2F6472636E6E6A2F736460656473","phone_udid":"A00407EF9D6EBCC207A514CDA452EB76","phone_udid2":"A00407EF9D6EBCC207A514CDA452EB76","appUdid":"1C35633F4EB8218789EFD8666C763485","phone_resolution":"2086*1080","phone_apn":"CMCC","phone_model":"ONEPLUSA6000","phone_firmware_version":"9","phone_softversion":"3.19.0","phone_softname":"com.esbook.reader","sdk_version":"3.1.8","cpid":"blf1298_12242_001","currentnetworktype":"4gnet","phone_city":"","os":"android","install_path":"\/data\/app\/com.esbook.readerTlgFCk6ANgEDRnXDCem8uQ==\/base.apk","last_cpid":"","package_name":"com.esbook.reader","src_code":"460001122544742"}
{"time":1550198433102,"type":"type_mobileinfo","phone_number":"15077113477","phone_imei":"860364049874919","phone_imsi":"460023771256711","phone_mac":"02:00:00:00:00:00","appkey":"307A5C626E6C2F6472636E6E6A2F736460656473","phone_udid":"EEF566CB5253AA62B653347A203815C3","phone_udid2":"0845931539AE39B3B0D4EB42B85D98EC","appUdid":"9570DCA2D574E6B69B24137035209D42","phone_resolution":"2340*1080","phone_apn":"CHINAMOBILE","phone_model":"PBEM00","phone_firmware_version":"8.1.0","phone_softversion":"3.19.0","phone_softname":"com.esbook.reader","sdk_version":"3.1.8","cpid":"blf1298_12242_001","currentnetworktype":"4gnet","phone_city":"","os":"android","install_path":"\/data\/app\/com.esbook.readerNBToXQo14TOeNuPxo_aA4w==\/base.apk","last_cpid":"","package_name":"com.esbook.reader","src_code":"13598c2d-efc4-4957-8d4d-22eb145d15fd"}
{"time":1550198440577,"type":"type_mobileinfo","phone_imei":"869800021106037","phone_imsi":"","phone_mac":"2c:5b:b8:fb:79:af","appkey":"307A5C626E6C2F6472636E6E6A2F736460656473","phone_udid":"2BC16C4AC07070BA9608BBD0EE2EE320","phone_udid2":"A7F9FA4772D31FADEECFDB445BA3BEBB","appUdid":"DC6BEE2F6E5D6A133E26131887AE788A","phone_resolution":"960*540","phone_apn":"","phone_model":"OPPOA33","phone_firmware_version":"5.1.1","phone_softversion":"3.19.0","phone_softname":"com.esbook.reader","sdk_version":"3.1.8","cpid":"blp1375_14526_003","currentnetworktype":"wifi","phone_city":"","os":"android","install_path":"\/data\/app\/com.esbook.reader1\/base.apk","last_cpid":"","package_name":"com.esbook.reader","src_code":"WIFIMAC:2c:5b:b8:fb:79:af"}
{"time":1506944701166,"type":"type_mobileinfo","phone_number":"+8618602699126","phone_imei":"865902038154143","phone_imsi":"460012690618403","phone_mac":"02:00:00:00:00:00","appkey":"307A5C626E6C2F6472636E6E6A2F736460656473","phone_udid":"388015DA70C0AEA6D59D3CE37B0C4BA2","phone_udid2":"388015DA70C0AEA6D59D3CE37B0C4BA2","appUdid":"EC0A105297D55075526018078A4A1B84","phone_resolution":"1920*1080","phone_apn":"中国联通","phone_model":"MIMAX2","phone_firmware_version":"7.1.1","phone_softversion":"3.19.0","phone_softname":"com.esbook.reader","sdk_version":"3.1.8","cpid":"blf1298_10928_001","currentnetworktype":"wifi","phone_city":"","os":"android","install_path":"\/data\/app\/com.esbook.reader1\/base.apk","last_cpid":"","package_name":"com.esbook.reader","src_code":"460012690618403"}
复制代码
7.执行 gpkafka 加载数据
[gpadmin@oracle166 ~]$ gpkafka load --quit-at-eof ./gpkafka_mobile_yamlPartitionID StartTime EndTime BeginOffset EndOffset0 2019-02-27T09:26:27.989312Z 2019-02-27T09:26:27.99517Z 0 5Job dcd0d159282c0ef39f182cabeef23ee6 stopped normally at 2019-02-27 09:26:29.442874281 +0000 UTC
复制代码
8.检查加载操作的进度(非必要)
[gpadmin@oracle166 ~]$ gpkafka check ./gpkafka_mobile_yamlPartitionID StartTime EndTime BeginOffset EndOffset0 2019-02-27T09:26:27.989312Z 2019-02-27T09:26:27.99517Z 0 5
复制代码
9.查看表数据
[gpadmin@oracle166 ~]$ psql
psql (8.3.23)
Type "help" for help.
lottu=# select * from tbl_novel_mobile_log ;
package_name | appkey | ts | phone_udid | os | idfa | phone_imei | cpid | last_cpid | p
hone_number
-------------------+------------------------------------------+---------------+----------------------------------+---------+------+-----------------+--
com.esbook.reader | 307A5C626E6C2F6472636E6E6A2F736460656473 | 1550198435941 | 8F137BFFB2289784A5EA2DCADCE519C2 | android | | 861738033581011 | blp1375_13621_001 | |
com.esbook.reader | 307A5C626E6C2F6472636E6E6A2F736460656473 | 1550198438311 | A00407EF9D6EBCC207A514CDA452EB76 | android | | 867520045576831 | blf1298_12242_001 | |
com.esbook.reader | 307A5C626E6C2F6472636E6E6A2F736460656473 | 1550198433102 | EEF566CB5253AA62B653347A203815C3 | android | | 860364049874919 | blf1298_12242_001 | | 1
5077113477
com.esbook.reader | 307A5C626E6C2F6472636E6E6A2F736460656473 | 1550198440577 | 2BC16C4AC07070BA9608BBD0EE2EE320 | android | | 869800021106037 | blp1375_14526_003 | |
cn.wejuan.reader | 307A5C626F2F76646B74606F2F736460656473 | 1550198437885 | A3BB70A0218AEFC7908B1D79C0C02D77 | android | | 862245038046551 | blf1298_14411_001 | |
(5 rows)
复制代码
三、后记
编辑本文初衷是：公司计划为北京 ES 小说作投放计划，需要类似热云数据平台作为投放数据支持，使投放更加精准可靠。北京小说部门数据存放于 kafka 中，需要将 kafka 中的数据导入深圳后台数据库中，虽然最后平台未采用 gpkafka 方式，但不失一种方案，由于种种原因后台数据库选 PG9.6 版本，采用 java 代码实现 kafka 数据实时导入 PG。最后祝 PG，GP 越来越好，也期待 pgkafka 工具诞生。
四、参考文献
1、gpkafka 更多用法
https://gpdb.docs.pivotal.io/5120/greenplum-kafka/intro.html
2、BottledWater-PG:PostgreSQL 集成 Kafka 的实时数据交换平台
https://www.jianshu.com/p/c3659f49bf94
作者介绍：
Lottu，就职于深圳宜搜科技有限公司，担任数据库 DBA，主要承担 PostgreSQL、Oracle 数据库维护工作以及数据库去 O 工作。
原文链接：
https://mp.weixin.qq.com/s/HuYYvKtV8RfNxtrrJwR6SQ
划线
评论
复制
更多内容推荐
基于 Kafka 技术栈构建和部署实时搜索引擎的实践
实现强大的搜索能力——从设计决策到幕后的一切
选型：不同阶段的数据应如何存储？
今天提到的MySQL、PostgreSQL、Redis、Doris、Etcd等，都是我们思考的一个具象化的表达而已。我们更关心的，应该是构建一个系统的思维。
2022-09-30
宜人贷 PaaS 数据服务平台简介（下篇）
在上篇中，我们已经简单了解了宜人贷数据平台Genie的特点，并且掌握了数据平台发展历程的一些信息。
Serverless ETL —— 蘑菇街实战落地
轻量、快速、灵活、省心、低成本的 Serverless 通用数据处理解决方案！
Spark Connecter：Cassandra
2020-12-17
62｜Apache Hive 集成
2020-11-12
银行数据库迁移至 MySQL，竟被时间字段这玩意耍了……
笔者在工作中曾遇到SyBase数据库迁移至MySQL时的一个问题：使用bcp将SyBase中的数据导出为csv文件时，datetime数据类型默认导出格式与MySQL不兼容。
kafka 核心技术与实战学习笔记（一）
kafka核心技术与实战 学习笔记
2021-08-29
OpenTelemetry：如何利用 OpenTelemetry 采集可观测数据？
这节课，我们来看看 OpenTelemetry ，它目前已经是可观测领域非常成熟和热门的标准和规范了。
2022-09-21
大数据总线平台 DBus 设计思路与工作原理
DBus专注于数据的收集及实时数据流计算，通过简单灵活的配置，以无侵入的方式对源端数据进行采集。
产品图鉴：哪些分布式数据库值得看？
今天这一讲，我会把视角切换到产品方向，为你做一次整体介绍。所以，你也可以将这一讲当作一个产品版的课程索引。
2020-10-16
使用 Kafka，如何成功迁移 SQL 数据库中超过 20 亿条记录？
不同的情况需要不同的解决方案，不过也许有人可以从我们的解决方案中得到一些有价值的见解。
数据库表结构变更处理方案
本篇文章主要介绍在DBus的设计中，它是如何处理表结构变更及其带来的各种问题的。
如何使用 Kafka、MongoDB 和 Maxwell’s Daemon 构建 SQL 数据库的审计系统
在本文中，作者讨论了在传统内置的数据副本之外，实现数据库审计日志系统的重要性，并给出了使用Kafka、MongoDB和Maxwell's Daemon构建审计日志系统的方案。
快速实现数据导入及简单 DCS 的实现
如何基于京东云数据计算服务平台，简单、快速地将数据导入数据计算服务。
spring-boot-route（十四）整合 Kafka
kafka简介及Spring Boot整合kafka
2020-10-13
使用数据湖工厂（DLF）搭建数据中心实时告警平台
一个数据中心部署了很多应用，需要建立统一的运维系统，实时接收应用的告警信息
数据库在各行各业的应用实践
如今，全社会驶入数字化转型快车道，数字生态正在推动着千行百业的变革。作为数字化转型的基石，数据库的重要性不言而喻。数据库更是成为企业的核心资产，运转和增长的驱动引擎，开发、运维等人员几乎每天都在和数据库接触。那么，在数据规模不断增长的今天，
2021-06-21
44｜技术演进（下）：软件架构和应用生命周期技术演进之路
今天，我会介绍下应用维度和应用生命周期管理维度的技术演进。
2021-09-04
超详细：这份全网首发的 Kafka 技术手册，从基础到实战一应俱全！
Kafka通常用于实时流数据架构，以提供实时分析。
2022-09-17
发现更多内容
推荐阅读
三天吃透 Kafka 面试八股文
2023-03-09
四次迭代，视源股份（CVTE）大规模实时流计算应用历程
实时计算03｜架构概述：一个监控系统的典型架构是什么样的？
2023-01-13
Kafka 基本原理、生产问题总结及性能优化实践 | 京东云技术团队
2023-11-01
百亿大表 Join 提速 300 倍！Apache Doris 在约苗数据平台的实时数仓建设实践
开篇词｜每个关注高可用的人，都应该了解监控知识
2023-01-09
25｜从集群角度拆解 Kafka 的架构设计与实现
2023-08-16
电子书
大厂实战PPT下载
换一换 
基于 Data Fabric 的逻辑数据湖架构实践
郭忆 | 网易 网易数帆 有数产品负责人
立即下载
将云原生进行到底：腾讯百万级别容器云平台实践揭秘
林沐 | 腾讯云 高级工程师
立即下载
Micro-frontends anti-patterns（微前端的反模式开发）
Luca Mezzalira | Amazon Web Service Principal Serverless Specialist Architect
立即下载
评论