我正在使用一个像以下架构一样生成架构的avro值变换（这只是一个子集，因为它很大）
{
"type": "record",
"name": "Envelope",
"namespace": "mssql.dbo.InvTR_T",
"fields": [
{
"name": "before",
"type": [
"null",
{
"type": "record",
"name": "Value",
"fields": [
{
"name": "InvTR_ID",
"type": "int"
},
{
"name": "Type_CH",
"type": "string"
},
{
"name": "CalcType_CH",
"type": "string"
},
{
"name": "ER_CST_ID",
"type": "int"
},
{
"name": "ER_REQ_ID",
"type": "int"
},
{
"name": "Vendor_ID",
"type": "int"
},
{
"name": "VendInv_VC",
"type": "string"
},
{
"name": "Status_CH",
"type": "string"
},
{
"name": "Stage_TI",
"type": {
"type": "int",
"connect.type": "int16"
}
},
{
"name": "CheckOut_ID",
"type": [
"null",
"int"
],
"default": null
},
{
"name": "ReCalcCk_LG",
"type": "boolean"
},
{
"name": "ReCalcAll_LG",
"type": "boolean"
},
{
"name": "PatMatch_LG",
"type": "boolean"
},
{
"name": "DocPatOvRd_LG",
"type": "boolean"
},
{
"name": "Locked_LG",
"type": [
"null",
"boolean"
],
"default": null
},
{
"name": "SegErrFlag_LG",
"type": "boolean"
},
{
"name": "Hold_LG",
"type": "boolean"
},
{
"name": "Reason_ID",
"type": [
"null",
{
"type": "int",
"connect.type": "int16"
}
],
"default": null
},
{
"name": "HoldCom_VC",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "AllSegFin_LG",
"type": "boolean"
},
{
"name": "InvAmt_MN",
"type": {
"type": "bytes",
"scale": 4,
"precision": 19,
"connect.version": 1,
"connect.parameters": {
"scale": "4",
"connect.decimal.precision": "19"
},
"connect.name": "org.apache.kafka.connect.data.Decimal",
"logicalType": "decimal"
}
当我运行以下以创建以下流程时
CREATE STREAM stream_invtr_t_json WITH (KAFKA_TOPIC='InvTR_T', VALUE_FORMAT='AVRO');
然后我描述了该流，该架构的格式非常奇怪。我想使用KSQL来过滤特定的信息并适当地分散这些事件。但是，我无法从kafka主题=> ksql stream => kafka主题=>接收器。如果我在该流中创建一个新主题，然后尝试将其消化到水槽中，我会得到
Expected Envelope for transformation, passing it unchanged
，然后丢失了PK的错误。我试图删除未包装转换，只是为了查看它的出现并在此上也收到错误。
BEFORE | STRUCT<INVTR_ID INTEGER, TYPE_CH VARCHAR(STRING), CALCTYPE_CH VARCHAR(STRING), ER_CST_ID INTEGER, ER_REQ_ID INTEGER, VENDOR_ID INTEGER, VENDINV_VC VARCHAR(STRING), STATUS_CH VARCHAR(STRING), STAGE_TI INTEGER, CHECKOUT_ID INTEGER, RECALCCK_LG BOOLEAN, RECALCALL_LG BOOLEAN, PATMATCH_LG BOOLEAN, DOCPATOVRD_LG BOOLEAN, LOCKED_LG BOOLEAN, SEGERRFLAG_LG BOOLEAN, HOLD_LG BOOLEAN, REASON_ID INTEGER, HOLDCOM_VC VARCHAR(STRING), ALLSEGFIN_LG BOOLEAN, INVDATE_DT BIGINT, SHIPDATE_DT BIGINT, PDTERMS_CH VARCHAR(STRING), PMTDUE_DT BIGINT, PMTTERMS_VC VARCHAR(STRING), BILLTERMS_CH VARCHAR(STRING), JOINT_LG BOOLEAN, COMMENT_VC VARCHAR(STRING), SOURCE_CH VARCHAR(STRING), ADDBY_ID VARCHAR(STRING), ADDED_DT BIGINT, CHGBY_ID VARCHAR(STRING), CHGED_DT BIGINT, APPROVED_LG BOOLEAN, MULTIPO_VC VARCHAR(STRING), PRVAUDITED_INVTR_ID INTEGER, PRVVENDOR_ID INTEGER, TRANSITDAYS_SI INTEGER, SHIP_NUM_VC VARCHAR(STRING), PRVTRANSITDAYS_SI INTEGER, PRVJOINT_LG BOOLEAN, CLONEDFROM_INVTR_ID INTEGER, LASTCALC_DT BIGINT, TMSFMANUAL_LG BOOLEAN, FRTRATERSOURCE_CH VARCHAR(STRING), ACTPICKUP_DT BIGINT, ROUTVEND_SI INTEGER, CALCVRSN_TI INTEGER, VENDORRANK_SI INTEGER, SEQ_SI INTEGER, PRVAUDITED_DT BIGINT, FRTRATERBATCHTYPE_CH VARCHAR(STRING), CURRENCY_TYPE_CD VARCHAR(STRING), EXCHANGE_DT BIGINT, EXCHANGE_RATE_LOCKED_LG BOOLEAN, EXCHANGE_DT_LOCKED_LG BOOLEAN, CUSTAPPROVED_LG BOOLEAN, FRTRATERMATCH_INVTR_ID INTEGER, CRC_INVOICE_LG BOOLEAN, RG_ROUTVEND_SI INTEGER, RG_PRVVE
似乎有关UnwrapFromEnvelope的评论解决了您问题的一部分。这只会使小数没有通过的零件。
查看连接器的文档：https：//debezium.io/documentation/reference/Reference/1.1/connectors/postgresql.html
我可以看到有一个decimal.handling.mode设置，如Jiri所说。在precise的默认值中，除非使用源数字或十进制类型，否则它将以ksqldb识别的格式输出AVRO小数。在这一点上，您最终使用了STRUCT数据结构，其中包括字节字段。
此规则有例外。当使用没有任何比例约束的数字或小数类型时，这意味着来自数据库的值对每个值具有不同的（变量）比例。在这种情况下
因此，要将数据导入KSQLDB，您将需要：
- 等到我们支持一个字节数据类型（目前尚未在我们的路线图上）
- 更改源表的模式以定义列的比例。
- 将DECIMAL.HANDLING.MODE更改为其他设置。您可能可以使用字符串，然后将值施放为KSQL中的小数。