CREATE TABLE DataInput (
     @timestamp VARCHAR,
     host       ROW<id VARCHAR,ip ARRAY<VARCHAR>>,
     log        ROW<offset INTEGER,file ROW<path VARCHAR>>,
     message    VARCHAR
 ) WITH (
     'connector' = 'kafka',   -- Optional: 'kafka', 'kafka-0.11'. Make sure to select the corresponding built-in Connector
     'topic' = 'topic-app-info',  -- Replace with the Topic you want to consume
     'scan.startup.mode' = 'earliest-offset', -- Can be any of latest-offset / earliest-offset / specific-offsets / group-offsets
     'properties.bootstrap.servers' = '10.0.0.29:9092',  -- Replace with your Kafka connection address
     'properties.group.id' = 'oceanus_group2',  -- Required parameter, make sure to specify a Group ID
     -- Definition Data formats (JSON formats)
     'format' = 'json',
     'json.ignore-parse-errors' = 'true',     -- Ignore JSON structure parsing exceptions
     'json.fail-on-missing-field' = 'false'   -- If set to true, an error will be thrown when a field is missing. If set to false, the missing field will be set to null
 );

CREATE TABLE es_output (
    id VARCHAR,
    ip ARRAY<VARCHAR>,
    path VARCHAR,
    num INTEGER,
    message VARCHAR,
    createTime VARCHAR
) WITH (
    'connector.type' = 'elasticsearch', -- Output to Elasticsearch
    'connector.version' = '6',          -- Specifies the version of Elasticsearch, e.g., '6', '7'. 
    'connector.hosts' = 'http://10.0.0.175:9200',  -- Elasticsearch connection address
    'connector.index' = 'oceanus_test2',       -- Name of the Elasticsearch Index
    'connector.document-type' = '_doc',  -- Type of the Elasticsearch document
    'connector.username' = 'elastic',  
    'connector.password' = 'yourpassword', 
    'update-mode' = 'upsert',            -- Optional 'append' mode without primary key or 'upsert' mode with primary key     
    'connector.key-delimiter' = '$',     -- Optional parameter, the delimiter for composite primary keys (defaults to '_', e.g., key1_key2_key3)
    'connector.key-null-literal' = 'n/a',  -- The substitute string when the primary key is null, defaults to 'null'
    'connector.failure-handler' = 'retry-rejected',   -- Optional error handling. Options are 'fail' (throw exception), 'ignore' (ignore any error), 'retry-rejected' (retry)

    'connector.flush-on-checkpoint' = 'true',   -- Optional parameter, disallows batch writes (flush) during snapshots, defaults to true
    'connector.bulk-flush.max-actions' = '42',  -- Optional parameter, maximum number of actions per batch
    'connector.bulk-flush.max-size' = '42 mb',  -- Optional parameter, maximum accumulated size per batch (supports only mb)
    'connector.bulk-flush.interval' = '60000',  -- Optional parameter, interval for batch writes (ms)
    'connector.connection-max-retry-timeout' = '1000',     -- Maximum timeout for each request (ms)
    --'connector.connection-path-prefix' = '/v1'          -- Optional field, the path prefix appended to each request                                                        
    'format.type' = 'json'        -- Output data format, currently supports only 'json'
);

INSERT INTO es_output
SELECT 
host.id as id,
host.ip as ip,
log.file.path as path,
log.offset as num,
message,
@timestamp as createTime
from DataInput;

# Query index. Replace username:password with the actual account password
curl -XGET -u username:password http://xx.xx.xx.xx:xxxx/oceanus_test2/_search -H 'Content-Type: application/json' -d'
{
    "query": { "match_all": {}},
    "size":  10
}
'