无法使用 MemSQL PIPELINE INTO PROCEDURE 摄取 JSON 数据答案

【问题标题】：Unable to ingest JSON data with MemSQL PIPELINE INTO PROCEDURE无法使用 MemSQL PIPELINE INTO PROCEDURE 摄取 JSON 数据
【发布时间】：2020-03-13 11:49:36
【问题描述】：

I am facing issue while ingesting a JSON data via PIPELINE to a table using Store Procedure.

I see NULL values are getting inserted in the table.

存储过程 SQL：

    DELIMITER //
    CREATE OR REPLACE PROCEDURE ops.process_users(GENERIC_BATCH query(GENERIC_JSON json)) AS
    BEGIN
        INSERT INTO ops.USER(USER_ID,USERNAME)
        SELECT GENERIC_JSON::USER_ID, GENERIC_JSON::USERNAME
        FROM GENERIC_BATCH;
    END //
    DELIMITER ;

使用的 MemSQL 管道命令：

    CREATE OR REPLACE PIPELINE ops.tweet_pipeline_with_sp AS LOAD DATA KAFKA ‘<KAFKA_SERVER_IP>:9092/user-topic’

    INTO PROCEDURE ops.process_users FORMAT JSON ;

推送到 Kafka 主题的 JSON 数据：{“USER_ID”:“111”,“USERNAME”:“Test_User”}

Table DDL Statement: CREATE TABLE ops.USER (USER_ID INTEGER, USERNAME VARCHAR(255));

【问题讨论】：

标签： apache-kafka singlestore

【解决方案1】：

您似乎在 https://www.memsql.com/forum/t/unable-to-ingest-json-data-with-pipeline-into-procedure/1702/3 的 MemSQL 论坛中获得帮助，尤其是 ::（生成 JSON）和 ::$（转换为 SQL 类型）之间的差异。

【讨论】：

【解决方案2】：

从 MemSQL 论坛获得解决方案！

以下是对我有用的管道和存储过程脚本，

CREATE OR REPLACE PIPELINE OPS.TEST_PIPELINE_WITH_SP 
AS LOAD DATA KAFKA '<KAFKA_SERVER_IP>/TEST-TOPIC'
INTO PROCEDURE OPS.PROCESS_USERS(GENERIC_JSON <- %) FORMAT JSON ;


DELIMITER //
CREATE OR REPLACE PROCEDURE ops.process_users(GENERIC_BATCH query(GENERIC_JSON json)) AS
BEGIN
INSERT INTO ops.USER(USER_ID,USERNAME)
SELECT GENERIC_JSON::USER_ID, json_extract_string(GENERIC_JSON,'USERNAME')
FROM GENERIC_BATCH;
END //
DELIMITER ;

【讨论】：