如何在 Apache Kafka 中通过 KSQL 分析 Twitter 数据

wudi

2017-11-09

介绍

KSQL 是 Apache Kafka 中的开源的流式 SQL 引擎。它可以让你在 Kafka 主题topic上，使用一个简单的并且是交互式的 SQL 接口，很容易地做一些复杂的流处理。在这个短文中，我们将看到如何轻松地配置并运行在一个沙箱中去探索它，并使用大家都喜欢的演示数据库源： Twitter。我们将从推文的原始流中获取，通过使用 KSQL 中的条件去过滤它，来构建一个聚合，如统计每个用户每小时的推文数量。

Confluent

如何在 Apache Kafka 中通过 KSQL 分析 Twitter 数据

首先，获取一个 Confluent 平台的副本。我使用的是 RPM 包，但是，如果你需要的话，你也可以使用 tar、 zip 等等。启动 Confluent 系统：

$ confluent start

（如果你感兴趣，这里有一个 Confluent 命令行的快速教程）

我们将使用 Kafka Connect 从 Twitter 上拉取数据。这个 Twitter 连接器可以在 GitHub 上找到。要安装它，像下面这样操作：

#Clone the git repo
cd/home/rmoff
gitclone https://github.com/jcustenborder/kafka-connect-twitter.git

#Compile the code
cd kafka-connect-twitter
mvn clean package

要让 Kafka Connect 去使用我们构建的连接器，你要去修改配置文件。因为我们使用 Confluent 命令行，真实的配置文件是在 etc/schema-registry/connect-avro-distributed.properties，因此去修改它并增加如下内容：

plugin.path=/home/rmoff/kafka-connect-twitter/target/kafka-connect-twitter-0.2-SNAPSHOT.tar.gz

重启动 Kafka Connect：

confluent stop connect
confluent start connect

一旦你安装好插件，你可以很容易地去配置它。你可以直接使用 Kafka Connect 的 REST API ，或者创建你的配置文件，这就是我要在这里做的。如果你需要全部的方法，请首先访问 Twitter 来获取你的 API 密钥。

{
"name":"twitter_source_json_01",
"config":{
"connector.class":"com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector",
"twitter.oauth.accessToken":"xxxx",
"twitter.oauth.consumerSecret":"xxxxx",
"twitter.oauth.consumerKey":"xxxx",
"twitter.oauth.accessTokenSecret":"xxxxx",
"kafka.delete.topic":"twitter_deletes_json_01",
"value.converter":"org.apache.kafka.connect.json.JsonConverter",
"key.converter":"org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable":false,
"key.converter.schemas.enable":false,
"kafka.status.topic":"twitter_json_01",
"process.deletes":true,
"filter.keywords":"rickastley,kafka,ksql,rmoff"
}
}

假设你写这些到 /home/rmoff/twitter-source.json，你可以现在运行：

$ confluent load twitter_source -d /home/rmoff/twitter-source.json

然后推文就从大家都喜欢的网络明星 [rick] 滚滚而来……

$ kafka-console-consumer --bootstrap-server localhost:9092--from-beginning --topic twitter_json_01|jq '.Text'
{
"string":"RT @rickastley: 30 years ago today I said I was Never Gonna Give You Up. I am a man of my word - Rick x https://t.co/VmbMQA6tQB"
}
{
"string":"RT @mariteg10: @rickastley @Carfestevent Wonderful Rick!!\nDo not forget Chile!!\nWe hope you get back someday!!\nHappy weekend for you!!\n❤…"
}

KSQL

现在我们从 KSQL 开始 ! 马上去下载并构建它：

cd/home/rmoff
gitclone https://github.com/confluentinc/ksql.git
cd/home/rmoff/ksql
mvn clean compile install -DskipTests

构建完成后，让我们来运行它：

./bin/ksql-cli local--bootstrap-server localhost:9092

======================================
= _ __ _____ ____ _ =
=||/ // ____|/ __ \| |=
=|' /| (___ | | | | | =
= | < \___ \| | | | | =
= | . \ ____) | |__| | |____ =
= |_|\_\_____/ \___\_\______| =
= =
= Streaming SQL Engine for Kafka =
Copyright 2017 Confluent Inc.
CLI v0.1, Server v0.1 located at http://localhost:9098
Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!
ksql>

使用 KSQL，我们可以让我们的数据保留在 Kafka 主题上并可以查询它。首先，我们需要去告诉 KSQL 主题上的数据模式schema是什么，一个 twitter 消息实际上是一个非常巨大的 JSON 对象，但是，为了简洁，我们只选出其中几行：

ksql> CREATE STREAM twitter_raw (CreatedAt BIGINT,Id BIGINT,Text VARCHAR) WITH (KAFKA_TOPIC='twitter_json_01', VALUE_FORMAT='JSON');
Message
----------------
Stream created

在定义的模式中，我们可以查询这些流。要让 KSQL 从该主题的开始展示数据（而不是默认的当前时间点），运行如下命令：

ksql> SET 'auto.offset.reset'='earliest';
Successfully changed localproperty'auto.offset.reset'from'null' to 'earliest'

现在，让我们看看这些数据，我们将使用 LIMIT 从句仅检索一行：

ksql> SELECT text FROM twitter_raw LIMIT 1;
RT @rickastley:30 years ago today I said I was NeverGonnaGiveYouUp. I am a man of my word -Rick x https://t.co/VmbMQA6tQB
LIMIT reached for the partition.
Query terminated
ksql>

现在，让我们使用刚刚定义和可用的推文内容的全部数据重新定义该流：

ksql> DROP stream twitter_raw;
Message
--------------------------------
Source TWITTER_RAW was dropped
ksql> CREATE STREAM twitter_raw (CreatedAt bigint,Id bigint,Text VARCHAR, SOURCE VARCHAR,Truncated VARCHAR,InReplyToStatusId VARCHAR,InReplyToUserId VARCHAR,InReplyToScreenName VARCHAR,GeoLocation VARCHAR,Place VARCHAR,Favorited VARCHAR,Retweeted VARCHAR,FavoriteCount VARCHAR,User VARCHAR,Retweet VARCHAR,Contributors VARCHAR,RetweetCount VARCHAR,RetweetedByMe VARCHAR,CurrentUserRetweetId VARCHAR,PossiblySensitive VARCHAR,Lang VARCHAR,WithheldInCountries VARCHAR,HashtagEntities VARCHAR,UserMentionEntities VARCHAR,MediaEntities VARCHAR,SymbolEntities VARCHAR,URLEntities VARCHAR) WITH (KAFKA_TOPIC='twitter_json_01',VALUE_FORMAT='JSON');
Message
----------------
Stream created
ksql>

现在，我们可以操作和检查更多的最近的数据，使用一般的 SQL 查询：

ksql> SELECT TIMESTAMPTOSTRING(CreatedAt,'yyyy-MM-dd HH:mm:ss.SSS') AS CreatedAt,\
EXTRACTJSONFIELD(user,'$.ScreenName')asScreenName,Text \
FROM twitter_raw \
WHERE LCASE(hashtagentities) LIKE '%oow%' OR \
LCASE(hashtagentities) LIKE '%ksql%';
2017-09-2913:59:58.000| rmoff |Looking forward to talking all about @apachekafka&@confluentinc’s #KSQL at #OOW17 on Sunday13:45 https://t.co/XbM4eIuzeG

注意这里没有 LIMIT 从句，因此，你将在屏幕上看到 “continuous query” 的结果。不像关系型数据表中返回一个确定数量结果的查询，一个持续查询会运行在无限的流式数据上，因此，它总是可能返回更多的记录。点击 Ctrl-C 去中断然后返回到 KSQL 提示符。在以上的查询中我们做了一些事情：

TIMESTAMPTOSTRING 将时间戳从 epoch 格式转换到人类可读格式。（LCTT 译注： epoch 指的是一个特定的时间 1970-01-01 00:00:00 UTC）
EXTRACTJSONFIELD 来展示数据源中嵌套的用户域中的一个字段，它看起来像：
1. {
2. "CreatedAt":1506570308000,
3. "Text":"RT @gwenshap: This is the best thing since partitioned bread :) https://t.co/1wbv3KwRM6",
4. [...]
5. "User":{
6. "Id":82564066,
7. "Name":"Robin Moffatt \uD83C\uDF7B\uD83C\uDFC3\uD83E\uDD53",
8. "ScreenName":"rmoff",
9. [...]
应用断言去展示内容，对 #（hashtag）使用模式匹配，使用 LCASE 去强制小写字母。（LCTT 译注：hashtag 是twitter 中用来标注线索主题的标签）

关于支持的函数列表，请查看 KSQL 文档。

我们可以创建一个从这个数据中得到的流：

ksql> CREATE STREAM twitter AS \
SELECT TIMESTAMPTOSTRING(CreatedAt,'yyyy-MM-dd HH:mm:ss.SSS') AS CreatedAt,\
EXTRACTJSONFIELD(user,'$.Name') AS user_Name,\
EXTRACTJSONFIELD(user,'$.ScreenName') AS user_ScreenName,\
EXTRACTJSONFIELD(user,'$.Location') AS user_Location,\
EXTRACTJSONFIELD(user,'$.Description') AS user_Description,\
Text,hashtagentities,lang \
FROM twitter_raw ;
Message
----------------------------
Stream created and running 
ksql> DESCRIBE twitter;
Field|Type
------------------------------------
ROWTIME | BIGINT 
ROWKEY | VARCHAR(STRING)
CREATEDAT | VARCHAR(STRING)
USER_NAME | VARCHAR(STRING)
USER_SCREENNAME | VARCHAR(STRING)
USER_LOCATION | VARCHAR(STRING)
USER_DESCRIPTION | VARCHAR(STRING)
TEXT | VARCHAR(STRING)
HASHTAGENTITIES | VARCHAR(STRING)
LANG | VARCHAR(STRING)
ksql>

并且查询这个得到的流：

ksql> SELECT CREATEDAT, USER_NAME, TEXT \
FROM TWITTER \
WHERE TEXT LIKE '%KSQL%';
2017-10-0323:39:37.000|NicolaFerraro| RT @flashdba:Again, I'm really taken with the possibilities opened up by @confluentinc's KSQL engine #Kafka https://t.co/aljnScgvvs

聚合

在我们结束之前，让我们去看一下怎么去做一些聚合。

ksql> SELECT user_screenname, COUNT(*) \
FROM twitter WINDOW TUMBLING (SIZE 1 HOUR) \
GROUP BY user_screenname HAVING COUNT(*)>1;
Oracleace |2
rojulman |2
smokeinpublic |2
ArtFlowMe|2
[...]

你将可能得到满屏幕的结果；这是因为 KSQL 在每次给定的时间窗口更新时实际发出聚合值。因为我们设置 KSQL 去读取在主题上的全部消息（SET 'auto.offset.reset' = 'earliest';），它是一次性读取这些所有的消息并计算聚合更新。这里有一个微妙之处值得去深入研究。我们的入站推文流正好就是一个流。但是，现有它不能创建聚合，我们实际上是创建了一个表。一个表是在给定时间点的给定键的值的一个快照。 KSQL 聚合数据基于消息的事件时间，并且如果它更新了，通过简单的相关窗口重申去操作后面到达的数据。困惑了吗？我希望没有，但是，让我们看一下，如果我们可以用这个例子去说明。我们将申明我们的聚合作为一个真实的表：

ksql> CREATE TABLE user_tweet_count AS \
SELECT user_screenname, count(*) AS tweet_count \
FROM twitter WINDOW TUMBLING (SIZE 1 HOUR) \
GROUP BY user_screenname ;
Message
---------------------------
Table created and running

看表中的列，这里除了我们要求的外，还有两个隐含列：

ksql> DESCRIBE user_tweet_count;
Field|Type
-----------------------------------
ROWTIME | BIGINT 
ROWKEY | VARCHAR(STRING)
USER_SCREENNAME | VARCHAR(STRING)
TWEET_COUNT | BIGINT 
ksql>

我们看一下这些是什么：

ksql> SELECT TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM-dd HH:mm:ss.SSS'), \
ROWKEY, USER_SCREENNAME, TWEET_COUNT \
FROM user_tweet_count \
WHERE USER_SCREENNAME='rmoff';
2017-09-2911:00:00.000| rmoff :Window{start=1506708000000end=-}| rmoff |2
2017-09-2912:00:00.000| rmoff :Window{start=1506711600000end=-}| rmoff |4
2017-09-2822:00:00.000| rmoff :Window{start=1506661200000end=-}| rmoff |2
2017-09-2909:00:00.000| rmoff :Window{start=1506700800000end=-}| rmoff |4
2017-09-2915:00:00.000| rmoff :Window{start=1506722400000end=-}| rmoff |2
2017-09-2913:00:00.000| rmoff :Window{start=1506715200000end=-}| rmoff |6

ROWTIME 是窗口开始时间， ROWKEY 是 GROUP BY（USER_SCREENNAME）加上窗口的组合。因此，我们可以通过创建另外一个衍生的表来整理一下：

ksql> CREATE TABLE USER_TWEET_COUNT_DISPLAY AS \
SELECT TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM-dd HH:mm:ss.SSS') AS WINDOW_START ,\
USER_SCREENNAME, TWEET_COUNT \
FROM user_tweet_count;
Message
---------------------------
Table created and running

现在它更易于查询和查看我们感兴趣的数据：

ksql> SELECT WINDOW_START , USER_SCREENNAME, TWEET_COUNT \
FROM USER_TWEET_COUNT_DISPLAY WHERE TWEET_COUNT>20;
2017-09-2912:00:00.000|VikasAatOracle|22
2017-09<

     kafka  apache  twitter

       
    wudi  
   0 关注   0 粉丝   0 动态  
 
    关注    关注   
 
  
    相关推荐 
      .NET Core下使用Kafka的方法步骤  
 
     # 修改每个topic的默认分区参数num.partitions，默认是1，具体合适的取值需要根据服务器配置进程确定，UCloud.ukafka = 3. Segment：partition物理上由多个segment组成，下面2.2和2.3有详细说明。pa     
 
   Kafka    2020-09-18  
 
      Apache Kafka 架构和相关概念  
 
     缓冲上下游瞬时突发流量，使其更平滑.特别是对于那种发送能力很强的上游系统，如果没有消息引擎的保护，“脆弱”的下游系统可能会直接被压垮导致全链路服务“雪崩”。使发送方与接收方松耦合,仅以协议的方式进行通讯,简化了开发.也叫消息队列,每个消息只能被一个下游的消     
 
   sweetgirl0    2020-07-27  
 
      Kafka、Flume、Kinesis更多的流方式支持，包括Twitter、ZeroMQ等  
 
     可线性伸缩至超过数百个节点；实现亚秒级延迟处理；可与Spark批处理和交互式处理无缝集成；提供简单的API实现复杂算法；更多的流方式支持，包括Kafka、Flume、Kinesis、Twitter、ZeroMQ等。Spark在接收到实时输入数据流后，将数据     
 
   jiaomrswang    2020-06-26  
 
      架构师的选择，Pulsar还是Kafka？  
 
     最近，我一直在研究Pulsar及其与Kafka的比较。快速搜索将显示两个最著名的开源消息传递系统之间存在当前的"战争"。作为Kafka的用户，我确实对Kafka的某些问题感到困惑，并且我对Pulsar感到非常失望。在本文中，我将重点介绍     
 
   yanghuashuiyue    2020-11-14  
 
      Kafka之消费与心跳  
 
     导读kafka是一个分布式，分区的，多副本的，多订阅者的消息发布订阅系统，可以用于搜索日志，监控日志，访问日志等。今天小编来领大家一起来学习一下Kafka消费与心跳机制。Kafka提供了非常简单的消费API，使用者只需初始化Kafka的Broker  Se     
 
   liuxingen    2020-11-13  
 
      刨根问底,Kafka消息中间件到底会不会丢消息  
 
     本文转载自微信公众号「爱笑的架构师」，作者雷架。大型互联网公司一般都会要求消息传递最大限度的不丢失，比如用户服务给代金券服务发送一个消息，如果消息丢失会造成用户未收到应得的代金券，最终用户会投诉。为避免上面类似情况的发生，除了做好补偿措施，更应该在系设计的     
 
   wangying    2020-11-13  
 
      记一次生产 Kafka 挂掉的那几分钟  
 
     本文转载自微信公众号「Java极客技术」，作者鸭血粉丝。Hello，大家好，我是阿粉，作为一个后端工程师不经历几次生产事故怎么能成长!阿粉工作几年来，大大小小，重要不重要的事故也经历了不少，有损失几十万的，有对业务毫无影响但是不应该发生的，每一次事故都是一     
 
   王谦    2020-11-03  
 
      面试 | 再有人问Kafka，别说你不会！  
 
     Kafka 是一个优秀的分布式消息中间件，许多系统中都会使用到 Kafka 来做消息通信。对分布式消息系统的了解和使用几乎成为一个后台开发人员必备的技能。今天就从常见的 Kafka 面试题入手和大家聊聊 Kafka 的那些事儿。生产者负责创建消息，然后将其     
 
   huangwei00    2020-10-14  
 
      RabbitMQ七战Kafka，差异立现！  
 
     作为一个有丰富经验的微服务系统架构师，经常有人问我，“应该选择RabbitMQ还是Kafka？”。基于某些原因， 许多开发者会把这两种技术当做等价的来看待。的确，在一些案例场景下选择RabbitMQ还是Kafka没什么差别，但是这两种技术在底层实现方面是有     
 
   shenzhenzsw    2020-10-09  
 
      Kafka突然宕机了？稳住，莫慌！  
 
     问题要从一次 Kafka 的宕机开始说起。笔者所在的是一家金融科技公司，但公司内部并没有采用在金融支付领域更为流行的  RabbitMQ，而是采用了设计之初就为日志处理而生的 Kafka，所以我一直很好奇 Kafka 的高可用实现和保障。从 Kafka 部     
 
   guicaizhou    2020-09-30  
 
      监控kafka队列长度  
 
     #!df=pd.DataFrame(columnList,columns=["topics","LAG"]). dfResult=df.groupby("topics",as_index=Fals     
 
   jiaomrswang    2020-09-23  
 
      Kafka都没整明白，还敢去面试？  
 
     放眼当下数据为王的时代，深入了解 Apache Kafka 及其常见的部署应用，快速实现数据架构已是大势所趋，刻不容缓。以下分别 Kafka 架构，四大核心 API，典型应用场景，Kafka 代理与消息主题，集群的创建，流 APIs及其处理模式等不同方面展     
 
   jyj0    2020-09-21  
 
      SAP Data Intelligence里的Kafka Producer和Consumer  
 
     新建一个graph，使用典型的生产者-消费者模型：将Data Generator生成的数据交给kafka Producer operator；而Kafka Consumer从kafka producer里读取出data Generator生成的数据，通过T     
 
   guicaizhou    2020-09-15  
 
      kafka技术题  
 
     将向 Kafka topic 发布消息的程序成为 producers.将预订 topics 并消费消息的程序成为 consumer.Kafka 以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个 broker.producers 通过网络将消息发     
 
   hannuotayouxi    2020-08-20  
 
      kafka常见面试题  
 
     我们举个例子说明下运维中面对的复杂性，我们都知道 kafka 有个 ISR集合，我先说明下这个概念：     
 
   yangyutong00    2020-08-09  
 
      Windows 安装kafka  
 
     直接去  D:\Tool\zookeeper-3.4.14\bin， 运行 zkServer.cmd. kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1      
 
   amwayy    2020-08-03  
 
      【kafka】消费组消费主题  
 
     当消费组中的消费者数量大于分区数量时，会存在浪费。因为有消费者分配不到partition进行消费。在消费者不饱和的情况下，一个消费者是可以去消费多个分区的数据的 。生产者生产投递9条消息，可以9条消息被投递到3不同分区，然后三个分区对应三个不同的消费者进行     
 
   yangyutong00    2020-08-01  
 
      logstash_output_kafka:Mysql同步Kafka深入详解  
 
     实际业务场景中，会遇到基础数据存在Mysql中，实时写入数据量比较大的情景。迁移至kafka是一种比较好的业务选型方案。如果需要同步历史全量数据+实时更新数据，建议使用logstash。实际上，核心logstash的同步原理的掌握，有助于大家理解类似的各种     
 
   weikaixxxxxx    2020-08-01  
 
      【kafka】生产者投递消息特征（异步发送）  
 
     kafka的生产者producer由持有未提交给kafka服务的记录的缓冲区构成一个缓冲区池，于此同时会有一个后台I/O线程来负责将这些记录record转化为请求request，然后将其提交给集群。如果发送完数据之后不对producer进行资源释放clos     
 
   PoppyEvan    2020-08-01  
 
      【kafka】消费者对应的分配partition分区策略  
 
     解决问题：1.实现了负载均衡。现有两个主题topic1、topic2，各自都有三个partition。TopicAndPartition对象，多个主题整体排序[topic1_partition0,topic1_partition1,topic1_parti     
 
   guicaizhou    2020-08-01