- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我使用 Confluent 4.1 对流和表中的数据进行了重新加密
1) 创建流
CREATE STREAM session_details_stream (Media varchar ,SessionIdTime varchar,SessionIdSeq long) with (kafka_topic = 'sessionDetails', value_format = 'json');
2) 创建重新加密的流,因为这个脚本不起作用,但在此之前它起作用了,为什么?
CREATE STREAM session_details_stream_rekeyed as select Media,SessionIdTime ,SessionIdSeq,CONCAT(SessionIdTime,SessionIdSeq) as root from SESSION_DETAILS_STREAM partition by root;
然后我创建下一个脚本s
CREATE STREAM session_details_stream_update as select Media,SessionIdTime ,SessionIdSeq,CONCAT(SessionIdTime,SessionIdSeq) as root from SESSION_DETAILS_STREAM partition by SessionIdTime;
CREATE STREAM session_details_stream_rekeyed as select Media,SessionIdTime ,SessionIdSeq,root from session_details_stream_update partition by root;
session_details_stream_rekeyed 的结果正常:
ksql> select * from session_details_stream_rekeyed;
1526411486488 | 2018-02-05T15:16:07.113+02:001| tex | 2018-02-05T15:16:07.113+02:001 | 1 | 2018-02-05T15:16:07.113+02:001
3) 为主题创建流;
CREATE STREAM voip_details_stream (SessionIdTime varchar,SessionIdSeq long) with (kafka_topic = 'voipDetails', value_format = 'json');
CREATE STREAM voip_details_stream_update as select SessionIdTime ,SessionIdSeq, CONCAT(SESSIONIDTIME,SESSIONIDSEQ) as root from voip_details_stream partition by SessionIdTime;
CREATE STREAM voip_details_stream_rekeyed6 as select SessionIdTime ,SessionIdSeq,root from voip_details_stream_update partition by root;
ksql> select * from voip_details_stream_rekeyed6;
1526411479438 | 2018-02-05T15:16:07.113+02:001 | 2018-02-05T15:16:07.113+02:00 | 1 | 2018-02-05T15:16:07.113+02:001
4) 创建表
CREATE TABLE voipDetails_table_test(SessionIdTime varchar,SessionIdSeq long,root varchar) WITH (kafka_topic='VOIP_DETAILS_STREAM_REKEYED6', value_format='JSON', KEY='root');
ksql> select * from voip_details_table;
1526411479438 | 2018-02-05T15:16:07.113+02:001 | 2018-02-05T15:16:07.113+02:00 | 1 | 2018-02-05T15:16:07.113+02:001
5) 然后我创建一个左连接
select c.root,u.root from session_details_stream_rekeyed c LEFT JOIN voipDetails_table_test u On c.root = u.root;
1526411477780 | 2018-02-05T15:16:07.113+02:001 | 2018-02-05T15:16:07.113+02:001 | null
问题出在哪里?
最佳答案
tl;dr 在进行流表连接时,您的表消息必须在流之前已经存在(并且必须带有时间戳)消息。如果您重新发送源流消息,则在填充表主题后,连接将成功。
使用kafkacat
填充主题(粘贴数据到stdin
)
cat > /tmp/msgs <<EOF
{"Media":"Foo","SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1}
{"Media":"Foo","SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2}
EOF
kafkacat -b localhost:9092 -P -t sessionDetails /tmp/msgs
cat > /tmp/msgs <<EOF
{"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1a"}
{"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1b"}
{"SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2,"Details":"Bar2"}
EOF
kafkacat -b localhost:9092 -P -t voipDetails /tmp/msgs
验证主题内容:
Robin@asgard02 ~> kafkacat -b localhost:9092 -C -t sessionDetails
{"Media":"Foo","SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1}
{"Media":"Foo","SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2}
Robin@asgard02 ~> kafkacat -b localhost:9092 -C -t voipDetails
{"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1a"}
{"SessionIdTime":"2018-05-17 11:25:33 BST","SessionIdSeq":1,"Details":"Bar1b"}
{"SessionIdTime":"2018-05-17 11:26:33 BST","SessionIdSeq":2,"Details":"Bar2"}
ksql> CREATE STREAM session_details_stream \
(Media varchar ,SessionIdTime varchar,SessionIdSeq long) \
WITH (KAFKA_TOPIC = 'sessionDetails', VALUE_FORMAT = 'json');
Message
----------------
Stream created
----------------
ksql> CREATE STREAM voip_details_stream \
(SessionIdTime varchar,SessionIdSeq long, Details varchar) \
WITH (KAFKA_TOPIC = 'voipDetails', VALUE_FORMAT = 'json');
Message
----------------
Stream created
----------------
ksql> select * from session_details_stream;
1526553130864 | null | Foo | 2018-05-17 11:25:33 BST | 1
1526553130865 | null | Foo | 2018-05-17 11:26:33 BST | 2
^CQuery terminated
ksql> select * from voip_details_stream;
1526553143176 | null | 2018-05-17 11:25:33 BST | 1 | Bar1a
1526553143176 | null | 2018-05-17 11:25:33 BST | 1 | Bar1b
1526553143176 | null | 2018-05-17 11:26:33 BST | 2 | Bar2
^CQuery terminated
ksql> CREATE STREAM SESSION AS \
SELECT Media, CONCAT(SessionIdTime,SessionIdSeq) AS root \
FROM session_details_stream \
PARTITION BY root;
Message
----------------------------
Stream created and running
----------------------------
ksql> SELECT ROWTIME, ROWKEY, root, media FROM SESSION;
1526553130864 | 2018-05-17 11:25:33 BST1 | 2018-05-17 11:25:33 BST1 | Foo
1526553130865 | 2018-05-17 11:26:33 BST2 | 2018-05-17 11:26:33 BST2 | Foo
ksql> CREATE STREAM VOIP AS \
SELECT CONCAT(SessionIdTime,SessionIdSeq) AS root, details \
FROM voip_details_stream \
PARTITION BY root;
Message
----------------------------
Stream created and running
----------------------------
ksql>
ksql> CREATE TABLE VOIP_TABLE (root VARCHAR, details VARCHAR) \
WITH (KAFKA_TOPIC='VOIP', VALUE_FORMAT='JSON', KEY='root');
Message
---------------
Table created
---------------
ksql> SELECT ROWTIME, ROWKEY, root, details FROM VOIP;
1526553143176 | 2018-05-17 11:26:33 BST2 | 2018-05-17 11:26:33 BST2 | Bar2
1526553143176 | 2018-05-17 11:25:33 BST1 | 2018-05-17 11:25:33 BST1 | Bar1a
1526553143176 | 2018-05-17 11:25:33 BST1 | 2018-05-17 11:25:33 BST1 | Bar1b
ksql> SELECT s.ROWTIME, s.root, s.media, v.details \
FROM SESSION s \
LEFT OUTER JOIN VOIP_TABLE v ON S.root = V.root;
1526553130864 | 2018-05-17 11:25:33 BST1 | Foo | null
1526553130865 | 2018-05-17 11:26:33 BST2 | Foo | null
让上面的 JOIN 查询保持运行。将 SESSION 消息重新发送到源主题(使用 kafkacat
将相同的消息发送到 sessionDetails
如上所述):
1526553862403 | 2018-05-17 11:25:33 BST1 | Foo | Bar1a
1526553988639 | 2018-05-17 11:26:33 BST2 | Foo | Bar2
Per Rohan Desai 在 Confluent Community Slack 上:
The problem is that the rowtime of the record from your stream is earlier than the rowtime of the record in your table that you expect it to join with. So when the stream record is processed there is no corresponding record in the table
使用 ROWTIME
查看源表上的消息以查看消息时间戳(不要与基于时间戳的 root
):
ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss') , ROWTIME, root, details from VOIP WHERE root='2018-05-17 11:26:33 BST2';
2018-05-17 11:32:23 | 1526553143176 | 2018-05-17 11:26:33 BST2 | Bar2
将此与源 session 流主题上的消息进行比较:
ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss') , ROWTIME, root, media from SESSION WHERE root='2018-05-17 11:26:33 BST2';
2018-05-17 11:32:10 | 1526553130865 | 2018-05-17 11:26:33 BST2 | Foo
2018-05-17 11:46:28 | 1526553988639 | 2018-05-17 11:26:33 BST2 | Foo
这些中的第一个(在11:32:10
/1526553130865
)先于相应的VOIP
消息(如上所示),并导致我们第一次看到的 null
连接结果。 second 这些日期较晚 (11:46:28
/1526553988639
) 产生了我们随后看到的成功连接:
1526553988639 | 2018-05-17 11:26:33 BST2 | Foo | Bar2
关于apache-kafka - Kafka ksql 简单连接不起作用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50371518/
我知道这个问题可能已经被问过,但我检查了所有这些,我认为我的情况有所不同(请友善)。所以我有两个数据集,第一个是测试数据集,第二个是我保存在数据框中的预测(预测值,这就是没有数据列的原因)。我想合并两
在 .loc 方法的帮助下,我根据同一数据框中另一列中的值来识别 Panda 数据框中某一列中的值。 下面给出了代码片段供您引用: var1 = output_df['Player'].loc[out
当我在 Windows 中使用 WinSCP 通过 Ubuntu 连接到 VMware 时,它提示: The server rejected SFTP connection, but it lis
我正在开发一个使用 xml web 服务的 android 应用程序。在 wi-fi 网络中连接时工作正常,但在 3G 网络中连接时失败(未找到 http 404)。 这不仅仅发生在设备中。为了进行测
我有一个XIB包含我的控件的文件,加载到 Interface Builder(Snow Leopard 上的 Xcode 4.0.2)中。 文件的所有者被设置为 someClassController
我在本地计算机上管理 MySQL 数据库,并通过运行以下程序通过 C 连接到它: #include #include #include int main(int argc, char** arg
我不知道为什么每次有人访问我网站上的页面时,都会打开一个与数据库的新连接。最终我到达了大约 300 并收到错误并且页面不再加载。我认为它应该工作的方式是,我将 maxIdle 设置为 30,这意味着
希望清理 NMEA GPS 中的 .txt 文件。我当前的代码如下。 deletes = ['$GPGGA', '$GPGSA', '$GPGSV', '$PSRF156', ] searchquer
我有一个 URL、一个用户名和一个密码。我想在 C# .Net WinForms 中建立 VPN 连接。 你能告诉我从哪里开始吗?任何第三方 API? 代码示例将受到高度赞赏... 最佳答案 您可以像
有没有更好的方法将字符串 vector 转换为字符 vector ,字符串之间的终止符为零。 因此,如果我有一个包含以下字符串的 vector "test","my","string",那么我想接收一
我正在编写一个库,它不断检查 android 设备的连接,并在设备连接、断开连接或互联网连接变慢时给出回调。 https://github.com/muddassir235/connection_ch
我的操作系统:Centos 7 + CLOUDLINUX 7.7当我尝试从服务器登录Mysql时 [root@server3 ~]# Mysql -u root -h localhost -P 330
我收到错误:Puma 发现此错误:无法打开到本地主机的 TCP 连接:9200(连接被拒绝 - 连接(2)用于“本地主机”端口 9200)(Faraday::ConnectionFailed)在我的
请给我一些解决以下错误的方法。 这是一个聊天应用....代码和错误如下:: conversations_controller.rb def create if Conversation.bet
我想将两个单元格中的数据连接到一个单元格中。我还想只组合那些具有相同 ID 的单元格。 任务 ID 名称 4355.2 参与者 4355.2 领袖 4462.1 在线 4462.1 快速 4597.1
我经常需要连接 TSQL 中的字段... 使用“+”运算符时 TSQL 强制您处理的两个问题是 Data Type Precedence和 NULL 值。 使用数据类型优先级,问题是转换错误。 1)
有没有在 iPad 或 iPhone 应用程序中使用 Facebook 连接。 这个想法是登录这个应用程序,然后能够看到我的哪些 facebook 用户也在使用该应用程序及其功能。 最佳答案 是的。
我在连接或打印字符串时遇到了一个奇怪的问题。我有一个 char * ,可以将其设置为字符串文字的几个值之一。 char *myStrLiteral = NULL; ... if(blah) myS
对于以下数据 - let $x := "Yahooooo !!!! Select one number - " let $y := 1 2 3 4 5 6 7 我想得到
我正在看 UDEMY for perl 的培训视频,但是视频不清晰,看起来有错误。 培训展示了如何使用以下示例连接 2 个字符串: #!usr/bin/perl print $str = "Hi";
我是一名优秀的程序员,十分优秀!