[
https://issues.apache.org/jira/browse/FLINK-19115?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17189012#comment-17189012
]
McClone commented on FLINK-19115:
---------------------------------
CREATE TEMPORARY FUNCTION UdfPrimaryKey as
'com.galaxy.sunny.flink.udf.UdfPrimaryKey' LANGUAGE JAVA;
CREATE TABLE my_kafka (
name STRING,
type STRING,
pass STRING
) WITH (
'connector' = 'kafka',
'topic' = 'my_data',
'properties.bootstrap.servers' = 'xxxxx:9092,xxxxx:9092',
'properties.group.id' = 'testGroup',
'format' = 'json',
'scan.startup.mode' = 'group-offsets'
);
CREATE TABLE my_es (
id STRING,
name STRING,
type STRING,
pass STRING,
primary key (id) not enforced
) WITH (
'connector' = 'elasticsearch-7',
'hosts'='http://xxxxx:9200',
'index'='flink-\{type}'
);
INSERT INTO my_es select UdfPrimaryKey(name,type,'2'),name,type,pass from
my_kafka;
----------------------------------------------------------------------------------------------------------------------
kafka json data:
{"name":"oYIuL","type":"6"}
es data:
{
"_index" : "flink-0",
"_type" : "_doc",
"_id" : "332a18f04899218f5d261870bc2e997e",
"_score" : 1.0,
"_source" : {
"id" : "332a18f04899218f5d261870bc2e997e",
"name" : "95ocK",
"type" : "0",
"pass" : null
}
},
> Null value fields cannot be ignored in ElasticSearch Sink
> ---------------------------------------------------------
>
> Key: FLINK-19115
> URL: https://issues.apache.org/jira/browse/FLINK-19115
> Project: Flink
> Issue Type: Bug
> Components: Connectors / ElasticSearch
> Affects Versions: 1.11.0
> Reporter: McClone
> Priority: Critical
>
> Null value fields cannot be ignored in ElasticSearch Sink
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
