我需要从其他九个主题的组合中创建一个Kafka主题,它们都是由Debezium PostgreSQL源连接器生成的,格式为AVRO。首先,我尝试(到目前为止还没有成功)组合来自两个主题的字段。
那么,首先基于"REQUEST"创建一个sqldb表主题:
ksql> CREATE TABLE TB_REQUEST (ID STRUCT<REQUEST_ID BIGINT> PRIMARY KEY)
WITH (KAFKA_TOPIC='REQUEST', FORMAT='AVRO');
对我来说一切都很好:
ksql> DESCRIBE TB_REQUEST;
Name : TB_REQUEST
Field | Type
-----------------------------------------------------------------------------------------------------------------------
ID | STRUCT<REQUEST_ID BIGINT> (primary key)
BEFORE | STRUCT<REQUEST_ID BIGINT, REQUESTER_ID INTEGER, STATUS_ID>
AFTER | STRUCT<REQUEST_ID BIGINT, REQUESTER_ID INTEGER, STATUS_ID>
SOURCE | STRUCT<VERSION VARCHAR(STRING), CONNECTOR VARCHAR(STRING), NAME VARCHAR(STRING), TS_MS BIGINT, SNAPSHOT VARCHAR(STRING), DB VARCHAR(STRING), SEQUENCE VARCHAR(STRING), SCHEMA VARCHAR(STRING), TABLE VARCHAR(STRING), TXID BIGINT, LSN BIGINT, XMIN BIGINT>
OP | VARCHAR(STRING)
TS_MS | BIGINT
TRANSACTION | STRUCT<ID VARCHAR(STRING), TOTAL_ORDER BIGINT, DATA_COLLECTION_ORDER BIGINT>
-----------------------------------------------------------------------------------------------------------------------
For runtime statistics and query details run: DESCRIBE <Stream,Table> EXTENDED;
然后从"EMPLOYEE"创建另一个表主题:
ksql> CREATE TABLE TB_EMPLOYEE (ID STRUCT<EMPLOYEE_ID INT> PRIMARY KEY)
WITH (KAFKA_TOPIC='EMPLOYEE', FORMAT='AVRO');
又一次,一切似乎都很好。
ksql> DESCRIBE TB_EMPLOYEE;
Name : TB_EMPLOYEE
Field | Type
-----------------------------------------------------------------------------------------------------------------------
ID | STRUCT<EMPLOYEE_ID INTEGER> (primary key)
BEFORE | STRUCT<EMPLOYEE_ID INTEGER, NAME VARCHAR(STRING), HIRING_DATE DATE>
AFTER | STRUCT<EMPLOYEE_ID INTEGER, NAME VARCHAR(STRING), HIRING_DATE DATE>
SOURCE | STRUCT<VERSION VARCHAR(STRING), CONNECTOR VARCHAR(STRING), NAME VARCHAR(STRING), TS_MS BIGINT, SNAPSHOT VARCHAR(STRING), DB VARCHAR(STRING), SEQUENCE VARCHAR(STRING), SCHEMA VARCHAR(STRING), TABLE VARCHAR(STRING), TXID BIGINT, LSN BIGINT, XMIN BIGINT>
OP | VARCHAR(STRING)
TS_MS | BIGINT
TRANSACTION | STRUCT<ID VARCHAR(STRING), TOTAL_ORDER BIGINT, DATA_COLLECTION_ORDER BIGINT>
-----------------------------------------------------------------------------------------------------------------------
For runtime statistics and query details run: DESCRIBE <Stream,Table> EXTENDED;
但是通过尝试创建我的目标表,通过Employee Id连接以前的表。
ksql> CREATE TABLE REQUEST_EMPLOYEE AS
SELECT RQ.ID->REQUEST_ID, RQ.AFTER->REQUESTER_ID, RQ.AFTER->STATUS_ID, EM.ID->EMPLOYEE_ID, EM.AFTER->NAME AS REQUESTER
FROM TB_REQUEST RQ
JOIN TB_EMPLOYEE EM ON RQ.AFTER->REQUESTER_ID = EM.ID->EMPLOYEE_ID;
我得到了以下错误:
Could not determine output schema for query due to error: Invalid join condition: table-table joins require to join on the primary key of the right input table. Got RQ.AFTER->REQUESTER_ID = EM.ID->EMPLOYEE_ID.
Statement: CREATE TABLE REQUEST_EMPLOYEE WITH (KAFKA_TOPIC='REQUEST_EMPLOYEE', PARTITIONS=1, REPLICAS=1) AS SELECT
RQ.ID->REQUEST_ID REQUEST_ID,
RQ.AFTER->REQUESTER_ID REQUESTER_ID,
RQ.AFTER->STATUS_ID STATUS_ID,
EM.ID->EMPLOYEE_ID EMPLOYEE_ID,
EM.AFTER->NAME REQUESTER
FROM TB_REQUEST RQ
INNER JOIN TB_EMPLOYEE EM ON ((RQ.AFTER->REQUESTER_ID = EM.ID->EMPLOYEE_ID))
EMIT CHANGES;
查看"DESCRIBE TB_EMPLOYEE"命令,在我看来&;EM.ID-> employee_id &;是正确的选择。我错过了什么?
提前感谢。
ksqlDB version in 0.21.0
我认为你应该在你的连接语句中使用至少一个行键,在以前的KsqlDB版本中,连接表的唯一方法是通过行键,在你当前的0.21.0版本中,可以使用外键。
检查以下示例:
CREATE TABLE orders_with_users AS
SELECT * FROM orders JOIN users ON orders.u_id = users.u_id EMIT CHANGES;
其中u_id被定义为主键,因此是rowkey。
CREATE TABLE users (
u_id VARCHAR PRIMARY KEY
name VARCHAR
) WITH (
kafka_topic = 'users',
partitions = 3,
value_format = 'json'
);
下面的句子类似
CREATE TABLE orders_with_users AS
SELECT * FROM orders JOIN users ON orders.u_id = users.ROWKEY EMIT CHANGES;
另一个观察结果是,sqldb认为tb_employee的键为STRUCT相关内容
最新更新
- 无法连接iPhone与xcode
- 正在创建globals()的副本
- 如何使用gstreamer从HLS流获取EXT-X-PROGRAM-DATE-TIME ?
- 如何在不刷新支柱的情况下测试salt - minion与saltmaster的连通性?
- 在创建后添加子对象到PSObject
- c - Gstreamer:如何直接从buffer_list创建视频|在Gstreamer中创建循环缓冲区
- 使用BFS算法遍历图后打印最短路径
- 如何在WINDOWS上使用公钥认证登录ssh-server ?
- 使用函数.在Woocommerce中,我如何隐藏缺货的简单产品?
- 在Xmonad中为特定窗口配置无边框
- 如何在aspx c#中获取客户端Ip地址或客户端/用户机器名
- 是否有可能在不使用GTM的情况下查看特定页面上点击了哪些链接?
- 根据年份 R 创建"时间段"索引变量
- R函数在管道中使用时不起作用
- React Native ios在Mac M1上构建失败
- pymysql.err.programmingError: (1064)
- Start返回0xE00002C7如果我的驱动程序继承IOUserUSBSerial
- 我如何传递一个特定的项目属性到一个变量在PowerShell?
- 基于SQL Server中某列的聚合,从单个记录生成多个记录
- 属性错误:模块"tensorflow.keras.callbacks"没有属性"BackupAndRestore"
- r语言 - 重命名数据框架而不将其写入全局环境
- 请求如何同时来自EC2实例和客户端机器?
- 使用ms图修改附件出现问题
- 使用laravel matwebsite只导入第二行的标题行
- 如何使用R Package将Tweedie GLM中的色散参数设置为1 (phi=1)
- 如何在c++中找到文件中不同组字符的总数
- PNG不显示在react tsx页面没有错误
- 如何实现导航元素的过渡效果?
- Use effect运行在依赖条件[count == 2]的下一个值检查上
- 实体框架更新一条记录删除另一条记录
热门标签:
javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium