PyFlink + 区块链?揭秘行业领头企业 BTC.com 如何实现实时计算

简介:实时计算的趋势,到流批一体的架构,通过对 PyFlink 和 Flink 的学习,稳定在线上运行了多种作业任务,对接了实际业务需求。并且搭建了 Zeppelin 平台,使得业务开发上更加方便。在计算上尽可能地依赖 SQL,方便各方面的集成与调试。

大家好,我们是 BTC.com 团队s V { f。2020 年,我们有幸接触到了 0 0 } I 8 ? Flink 和 PyFliN I Snk 生态,从团队自身需求出发,o Z w完善了团队内实时{ o , 3算的任务和需求,搭建了流批一体的计算环境。

在实现实时计算的过程中,我们在实践中收获了一些经验,在此分享一些这方面的心路历程。主要分享的大纲如下:

  • 困惑 • 描述 • 思考 • 行动
  • 流批一体的D ) + ;架构
  • 架构
  • 效果
  • Zeppelin、PyFlink on K8S 等实践
  • Zeppelin
  • PyFlink on K8S
  • 区块链领域实践
  • H 1 / T :望 • 总结

01 困惑 • 描述 • 思考 • 行V P Y 9 Z C q

作为工程师,我们每天都在不断地了解需求,研发业务。

有一天,我们被拉到了一次团队总结会议上,收到了以下的需求:

销售总] ? { K g监 A:

我们想要知道销售的历史和实时转H Q T 化率、销售额,能不能统计一下实时的 TOP5 的商品,还有就是大促时候,用户实时访问、商品实时浏览量 TOP5 的情况呢,可以根据他历史访问的记录实时推荐相关的吗?

市场总监 B:

我们想要知道市场推广的效果,每次活动的实时数据,不然我们的市场投放无法准确评估效果,及时反馈啊。

研发总监 C:

有些用户的 Bug 无法复现,日志可以再实时一点吗?传统日志分析,需要一定的梳理,可不可以直接清洗 / 处理相关的数据?

采购总监 D:

这些年是不是流行数字化,采购N i 4 ^这边想预测采购需求,做一下实时分类和管理支出,预测未来供应来源,完善一下成本。这个有办法做吗?还有有些供应商不太稳定啊,能监控到他T R u Q ? T p V b们的情况吗?

运维总监 E:

网站有时候访问比较慢,没有地方l 7 ?可以看到实时的机器情况,搞个什么监控大屏,这个有办法解决吗?

部门领导 F:

可以实现上面的人的需求吗。

做以上的了解之后,才发现,大家对于数据需求的渴望程度,使用方不仅需要历史的数据,而且还需要实时性的数据。

在电商、金融、制造等% . Y D行业,数据有着7 9 Z t迅猛的增长,诸多的企业面临着新的挑战,数据分析的实时处理框架,比如说做一些实时数据分析报表、实时数据处理计算等。

和大多数企业类似,在此之前,7 ! N N我们是没有实时计算这方面的经验和积累的。这时,就开始 h y ^ W困惑了,怎样可以更好地做上面的需求,在成本和效果之间取得平衡,如何设计相关的架构?

PyFlink + 区块链?揭秘行业领头企业 BTC.com 如何实现实时计算

穷则思变` = 2 Zv S + 2 j N m在有了困惑以后,我们就开始准备梳理已有的条件和我们到底需要什么。

我们的业务范围主要在区块链浏览器与数d i 8 z F | C据服务、区块链矿池、多币种钱包等。在区块链浏览器的业务里,BTC.com 目[ x J A R V X前已是全球领先的区块链数据服务平台,矿池业务在业内排行第一,区块链浏览器也是全球前三大浏览器之一。

首先,我们通过 parser 解析区块链上的数据,得: a Q A T ; % 1到各方面的数据信息w h # l @ Q /,可以分析出每个币种的地址活跃度、地址交易情况、交易流向、参与程度等内容。目前,BTC.com 区块链浏览器与行业内各大矿池和交易所等公司都有相关合作,可以更好地实现一些数据的统计、整理、归纳、输出等。

面向的用户,不仅有专业的区块链开发人员,也有各样的 B 端和 C 端用户,C 端用户可以进行区块链地址的标注,智能合约T 8 e的运行,查看智能合约相关内容等,以及链上数据的检索和查看。Bs 7 4 端用户则有更专业的支持和指导,提供 API、区块链节点等一些的定制以及交易加速、链上的业务合作、数据定制等。

从数据量级来讲,截至目前,比特币大概有 5 亿笔交易,; b T P Y3000 多万地址,22 亿输出(output:每笔交易的输出),并且还在不断增长中。以太坊的话,则更多。而 BTC.com 的Y a = x c矿池和区块链浏览器都支持I 0 P o d 5 L e N多币种,各币种的总数据量级约为几十 T。

矿池是矿工购买矿机设备后连接到的服务平台,矿工可以通过连接矿池从而获取更稳定的收益。这是一个需要保证 7 * 24 小时稳定的服务,里面有矿机不断地提交其计算好的矿池下发的任务的解,矿池将达到网络难度的解进行广播。这个过程也可以认为是近乎是实时的,矿机通过提交到服务器,服务器内部再提交到 Kafl ~ c & 3 F tka 消息队列,同时有一些组件监听这些消息进行消费。而这些提交上来的解可以从中分析出矿机的工作状态、- ] 8 z *算力、连接情况等。

在业务上,我们需要进行历史数据和实时数据的计算。

历史数据要关联一些币价,历史交易信息,而这些交易e 6 s l { @ t信息需要一直保存,是一种典型的批处理任务。

每当有新区块的确认,就有一些数据可以得B $ R f } + X l 7到处理和分析5 m M # O,比如某个地址在这个区块里发生了一笔交易,那么可以从其交易流向去分析是什么样的交易,挖掘交易相关性。或者是在这个区块里有一些特殊的交易,比如 segwit 的交易、比如闪电网络的交易,就j p p e 9是有一些这个币种特有的东西可以进Q B 4 @ ` Q行解析分析和统计。并且在新区块确认时的难度预测也有所变化

还有就是大额交易的监控,通过新区块的确认和未确认交易,锁定一些大额交易,结合地址的一些标注,锁定交易流向,更好地进行数据分析。

还有是一些区块链方面的 OLy Y l $ R e .AP 方面的需求。

PyFlink + 区块链?揭秘行业领头企业 BTC.com 如何实现实时计算

总结了在数据统计方面的需求和问题以后,我们就开L % _ | _ %始进行思考:什么是最合适的架构,如何让人员参与少、成本低?

解决问题,无非就是提出假设,通Q g . ]过度量,然后刷新认知。

PyFlink + 区块链?揭秘行业领头企业 BTC.com 如何实现实时计算

在浏览了一些资料以后,我们认为,大部分的计算框架都是通过输入,进行处理,然K ; { M ^ 2 ? f ^后得到输出。首先,我们要获取到数据,这里数据可以从 MySQL 也可以从 Kafka,然后进行计算,这里计算可以是聚合,也可以是 TOP 5 类型的,在实时的话,可能还会有窗口类型的。X , K P j a在计算u x { +完之后,将结果做( U H |下发,下发到消息渠道和存储,发送到微信或者钉钉,落9 8 3 d ~ ? 9 S地到 MySQL 等。

团队一开始尝试了 Spark,搭建了 Yarn,使用了 Airflow 作为调度框架,通过做 MySQL 的集成导入,开发了一些批处理任务,有着离线任务的特点,数据固定、量大、计算周期长,需要做一些复杂操作。

在一些批处理任务上w - & 9 %,这种架构是稳定的,但是随着业务的发展,有了越Y ~ M来越多的实时的需求,并且实时的数据并不能保证按顺序到达,按时间戳排序,消息的时间字段是允许前后有差距的。在数据模型上,4 4 U H 8 g ? H Z需求驱动式的开发,成本相对来说,Sp5 6 h N - D A ? 7ark 的方式对于当时来说较高,对于状态的处理不是很好,导致影响一部分的9 - g : Z效率。

其实在 2019 年的时候,就有在调研一些实时计算的事情,关注到了 Flink 框架,当时还是以 Java 为主,整体框架概念上和 Spark 不! i i D 0 E z同,认为批4 x i d处理是一种特殊的流,但是因为团队没有 Java 方面的基因和沉淀,使用 Flink 作为实时计算的架构,在当时就暂告一个段落。

在 2020 年初的时候,不管是 Fli U c 1 H ;nk 社区[ Y 4 s [ 5 还是I H V b = X # g InfoQ,还是 B 站,都有在推广 PyFlink,而且当时尤其是程鹤群[1]和孙金城[2]的视频以及孙金城老师的博客[3]的印象深刻。于是就想尝试 PyF` O e | p Slink,其有着流批一体的优势,而且还支持 Python 的一些函数,支持 pandas,甚至以后还可以支持 T[ e | * * w |ensorflow、Keras,这对Y f 7 | a M我们的& L - g吸引力是巨大的。在之后,就在构思我们的在 PyFlink 上的流批一体的架构。

02 流批一体的架构

架构

首先我们要梳理数据,要清楚数据从哪里来。在以 Spari p - #k 为主的时期,数据是定期从数据源加载(增量)数据,通过一定的转换逻辑,然后写入目的地,由6 B 5 p + 8于数据量和业务需要,延迟通常在小时级别,而实时的话,需要尽可能短的延迟,因此将数据源进行了分类,整体分成= + I ? Z _ A 6了几部分,一部分是传统的数据我们存放在 MySQL 持久化做保存,这; B e h B n )部分之后可以直接作为批处理的计算,也可以导入 Hive,f f K做进一步的计算。实时的部分,实际上是有很多思路,一种方式是通过 MySQL 的 Binlog 做解析,还有就是 MySQL 的 CDC 功能,在多方考量下,最后我们选择了 Kafka,不仅是因为其是优秀的分布式流式平台,而且团队也有对其的技术沉淀。

并且实际上在6 P D X C U本地开发的时候,安装 Kafka 也比较方便,只需要 Brew Install Kafka,而且通) 7 v X 0 O p过 ConS 2 u I j uduktor 客户端,也= # t C v R可以方便t Z 3 l C的看到每个 Topic 的情况。于是就对现有的 Parser 进行改造,使其支持 Kafka,在当收到新的区块时,会立即向 Kafka 发送一个消息,然后进行处理。

大概是在 2018 年的时候,团队将整体的业务迁移到了 Kubernetes 上,在业务不断发展的过程中,其对开发和运维上来说,减轻了很多负担,所以建议有一定规模的业务,最好是迁移到 Kubernetes,其对成本的优化,DevOps,以及高可用的支持,都是其他平台和传统方式无法比拟的。

在开发作业的过程中,我们在尽可能的使用 Flink SQL,同时结合一些 Java、Python 的 UDF、UDAF、UDTF。每个作业通过初始化类似于以下的语句,形成一定的模式:

self.source_ddl = '''

CREATE TABLE SourceTable (xxx int) WITH

'''

self.sink_ddl = '''

CREATE TABLE SinkTable (xxx int) WITH

'''

self.transform_ddl = '''

INSERTH b d INTO SinkTable

SELECT udf(xxx)

FROM SourceTable

GROUP BY FROM_UNIXN Q * T Z J 9TIMEo F b 4 %(`timestamp`, 'yyyyMMdd7 L I : J')

'''

在未来的话,会针对性地将数据进行分层,按照业界通用的 ODS、DW9 7 J I MD、DWS、ADS,分出原始层,明细层和汇总N h o u 9 U ? [ }层,进一步做好数据的治理。

效果

最终我们团队基于 PyFlink 开发快速地完成了已有的任务,部分是批处理作业,处理过去几天的数据,部分是实时作业,根据 Kafka 的消息进行消费,目前还算比较稳定。

z / F G e P e Q署时选择了 Kubernetes,具体下面会进行+ $ { D V S分享。在 K8S 部署了 Jobmanager 和 Taskmanager,并且使用 Kubernetes 的 job 功能作为批处理作业的部署,之后考虑接入一些监控平台,比如 Prometheus 之类的。

在成本方面,由于是使用的 Kubernetes 集群,因此在机器上只有扩展主机的成本,在这种方式上,成本要比传统的 Ya- . J ?rn 部署方式要低,并且之后 Kuberntes 会支持原生部署,在扩展 Jobmanager 和 Taskmanager 上面会更加方便。

039 G v R Q E Zeppelin、PyFlink on K8S 等实践

Zeppelin 是我们用来进行数据探索和逻辑验证,有些数据在本地不是真实数据,利用 Zeppelin 连接实际的链上数据,进行计算的逻辑验证,当验证完成后,便可转换成生产需要M d * 1 ] I %的代码进行部署。Q - } l Q Y % 6 I

PyFlink + 区块链?揭秘行业领头企业 BTC.com 如何实现实时计算

一、Kubernetes 上搭建 PyFlink 和 Z` S y u ? E Meppelin

1. 整理后的部署 Demo 在 github,可以参阅相关链接[4]。 2. 关于配置文件,修改以下配置的作用。

(1). 修改 configmap 的 flink-conf.yaml 文件的 taskmanager 配置。

taskmanager.numberOfTaskSlots: 10

调整 Taskmanager 可以# j l . I s调整运行的 job 的数量。

(2). 在 Zeppelin 的 dockerfile 中修改 zeppeliO [ Hn-site.xml 文件。

cp conf/zeppelin-site.xml.template conf/zeppelin-site.xml; \

sed -i 's#<value>127.0.0.1</value>#<value>0.0.0.0</value>#g' conf/zeppelin-site.xml; \

sed -i 's#<value>auto</value>#@ _ $ ^ & u S<value>local</v1 1 ualue&M l I G L - I ! %gt;#g' conf/zeppelin-site.xml

  • 修改请求来源为 0.0.0.0,如果是线上环境,建议开启白名单,加上 auth 认证。
  • 修改 interpreter 的启动模式为 local,auto 会导致在集群启动时,以 K8s 的模式启动,目前 K8s 模式只支持 Spark,local 模式可以理解为,Zeppelin 将在本地启动一个连接 Flink 的 interpreter 进程。
  • Zeppelin 和在本地提交 Flink 作业类似,也需要 PyFlink 的基k t t 8 _ [ c [础环境,所以a O ) 3需要将 Flink 对应版本的 jar 包放入镜像内。

3. Zeppelin 的 ingress 中添加 websocketv p F J @ 配置。

nginx.ingresa X q ` w 1s.kubernetes.io/configuration-snippet: |

proxyH B }_set_header UpgradeF & T Z 5 "websocket";

proxy_set_header Connection "Upgrade";

Zeppelin 在浏览器需要和 server 端建立 socket 连接,需要在 ingress 添加 websocket 配置。

4.Flink 和 Zeppelin 数据持久化的作用。

volumeMounts:

- mountPath: /zeppeliw d ! 6 )n/notebook/

name: data

volumes:

- name: data

persistentVolumeClaim:

claimName: zeppelin-pvc

--- a M B i A { /

apiVersion: v1

kind: PersistentVolumeClaim

metadata:

name: zeppelin-pvc

spec:

storageClassName: efs-sc

acce7 ! O ( l v ^ # OssModes:

- ReadWriteOnce

resources:

requests:N 8 + o C q

storage^ R 8 D: 1Gi

  • 对 Flink 的 /opt/flink/lib 目录做持久化的o & { p W g g目的,是当我们需要新的 jar 包时,可以直接进入 Flink 的 pod 进行下载,并存放到 li{ Q ; S $ w o ,b 目录,保证 jobmanager 和 taskmanager 的 jar 版本一致,同时也无需. @ y z 7 ] l更换镜像。
  • Zeppel{ _ F Din 的任务作业代码会存[ ; t放在 /zeppelin/notebook/ 目录下,目的是方便保存编写好的代码。

5. Flink 命令提交 job 作业的方式。

(1). 本地安装 PyFlink,Python 需要O F j u H u $3.5及以上版本。

$ pip3 install apacheJ ) { / c u z 7-flink==1.11.1

(2). 测试 Demo

def word_count():

env = StreamExecutionEnvironment.get_execution_environment()

t_env% s w L = StreamTableEnvironment.create(

env,

environment_settings=EnvironmentSettings.new_instance().use6 ^ E_blink_planner().build()

sink_ddl = """

create table Results (word VARCHAR, `count` BIGINT) with ( 'connector' = 'print')

"""

t_env.sq3 ` / 4 z = L T !l_update(sink_ddl)

elements = [(word, 1) for word in content.split(" ")]

# 这里也可以通过 Flink SQL

t_env.from_elements(elements, ["word", "count"]) \

.group_by("wor4 ! ]d") \

.selL [ y Oect("word, count(1) as count& ! 3 /") \

.insert_; S - r q { |into("Results")

t_env.execute(O x X @"w s 9 3ord_count")

if __name__ == '__main__'5 R ~ & k =:

logging.basicConfig(stream=sys.stdout,y g o k D _ 9 l* _ ) F l q C )evel=logging.INFO, format="%(message)s")

word_count()

或者是实时处理的 Demo:

def handle_kafka_message():

s_env = StreamExecutionEnvironment.get_execution_environment()

# s_env.set_stream_time_cha7 / A V .racteristic(TimeCharacY l N Zteristic.EventTime)

s_env.sB N k } C &et_parallelism(1)

st_env = StreT r X u @ 5 g VamTableEnvironment \

.create(s_env, environment_settings=EnvironmentSettings

.new_instance()

.in_streaming_mode()

.use_blink_plan$ P S w 1 9 h 1 #ner().build())

source_ddl = '''

CREATE TABLE Ss D g s Y W 6 * .ourc? I S % ?eTable (

wo3 ! 0rd string

) WITH (

'connector.type' = 'kafka',

'connector.version' = 'universal',

'connector.topic' = 'Topic',

'connector.properties.boot# ? % G _strap.servers' = 'localhostG 5 b } & R:9092',

'connector.properties.P ! % ~ Ezookeeper.connect' = 'd , ` A o 2 ~ d mlocalhost:2121',

'format.type' = 'json',

'format.derive-schema' = 'truef 9 * 9 G S ~'

'''

sink_dz _ ; cdl = """

create table Results (word Vo # . P } }ARCHAR) with ('connes V W [ 9 Actor' = 'print')

"""

st_env.sz % A ) ( Zql_update(sink_ N - r L ] ]ddl)

sR , R ) Y Y ,t_env.sql_update(source_ddl)

st_e} E p s ] inv.from_path("source").insert_into("sink")

st_env.execute("KafkaTest")

if __name__ == '__main5 i ^ 0 7 M l__':

handle_k: % # L 5afka_message()

(3). 本地测试 FliD 9 N M : ; $ Snk 命令提交 job 作业。

$ flink run -m localhost:8081 -py word_count.py

python/table/batch/w) L & Mord_count.py

Job has been submitted with JobID 0a31f a D ) n 7 h 9 !b61c2f974bcc3f34o c Q4f57829fc5d5

Program exet 3 Z A Z acution finished

Job with JobID 0a31+ d t Q lb61c2f974bcc3f344M V C $ `f579 q E ^ 6829fc5d5 has finished.

Job Rv d C a 1 b { j Runtime: 741 ms

(4). 如果存在多个 Python 文件,可以先 zip 打包后再进行提交作业。f $ B T 2 [ K

$ zip -r flinkdemo.zip ./*

$ flink run -m localhost:8081 -pyfs flinkdemo.zip -pym main

(5). Kubernetes 通过集群h Q 5 d k 2 C : A的 CronJob 定时调度来提交 Job,之后会做自研一些 UI 后台界面做作业管理与监控。

04 在区块链领域实践

随着区块链技术的越来越成熟,应用越来越多,行业标准化、规范化的趋势也开始显现,也越来越依W l H Z I赖于云计算、大数据,毕竟是数字经济的产物。BTC.com 也在扎根于区块链技术基础设施,为各类公司各类应用提供数据和业务上的支持。

近些年,有个词火遍了 IT 业界,中台,不管是大公司还是创业公司,都喜欢扯上@ K g i G )这个概念,号称自己业务中台,数据中台等。我们的理解中,中台是一种整合各方面资源的能力,从传统的单兵7 a c & : } v j作战,到提升武器装备后勤保障,提升作战能力。在数据上打破数据孤岛,在需求快速变化| w 2 -的前台和日趋稳定q h O X g b ` z的后台中取得z T n - v 7 J平衡。而中台更重要的是服务,最终g V F 2还是要回馈到客户,回馈到合作伙伴。

在区块链领域,BTC.com 有着深厚的行业技术积累,可以提供各方面数据化的能力。比如在利用机器学习进行链上数据的预估,预估 eth 的 gas price,还有最佳手续费等,利用 keras 深度学习的能力,进行一些回归计算,在之后也会将 Flink、机器学习和区块链结合起来,对外提供3 x H : 7 ] q更多预测类和规范化分类的数据样本,之前是在用定时任务不断训练模型,与 Flink 结合之后,会更加实时。在这方面,以后也会提供更多的课题,+ ? 8 l比如币价与 Defi,舆情,市场等的关系,区块链地址与交易的标注和分类。甚至于将机器学习训练的模型,放于 IPFS 网络中,通过去中心化的代币进行训练,提供方便调用样本, , & R h ? N R和模型的能力。

在目t E p s前,BTC.com 推出了一些通过数据挖掘实现的能力,包括交易推送、OLAP 链上分析报表等,改善和提升相关行业和开发者实际的体验。我们在各种链上都有监控节点,监控各区块链网络的可用性、去中心化程度,监控智能合约。在接入一些联盟链、隐私加密货币,可以为联盟链、隐私加密货币+ + + 0 r Y提供这方面的数据能力。

BTC.com 将为区块链产业生态发展做出更多努力,以科技公司的本质,以技术发展为第一驱动力,以市场和客户为导向,开发创新和融合应用,做好基础设施。

05 展望与总结

从实时计算的趋势,到流批一体的架构,通过对 PyFlink 和 Flink 的学习,稳定在线上运行了多种x 8 r 4 C Z _ I A作业任务,对接了实际业务需求。并且搭e ! & y ? ! b建了 Zeppelin 平台,使得业务开发上更加方便。在计算上尽可能地依赖 SQL,方便各方面的集成与调试。

在社区方面,PyFlil K ] ( | _ d !nk 也是没有令我们失望的,较快的响应能力,不断完善的文档。在 Cona = )fluence[5]上也可以看到一些 Flink Improvement Proposals,其中也有一些是 PyFlink 相关的,在不远L S J ( @ a t U W的将来,还会支持 Pandas UDAF,DataStream API,ML API,也期望在之后可以支持 Joblistener,总L + ^ ) d Y q g之,在* C . p C m这里也非常感谢相关团队。

未来的展望,总结起来就是,通过业务实现数据的价值化。而数据中台的终局,是将数据变现。

作者:阿里云实时计算Flink

本文为阿里云原创内容,未经允许不得转载。