赞
踩
以下示例展示了如何创建 Elasticsearch sink 表:
CREATE TABLE myUserTable (
user_id STRING,
user_name STRING,
uv BIGINT,
pv BIGINT,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'elasticsearch-7',
'hosts' = 'http://localhost:9200',
'index' = 'users'
);
参数 | 是否必选 | 默认值 | 数据类型 | 描述 |
---|---|---|---|---|
connector | 必选 | (none) | String | 指定要使用的连接器,有效值为:elasticsearch-6:连接到 Elasticsearch 6.x 的集群。elasticsearch-7:连接到 Elasticsearch 7.x 及更高版本的集群。 |
hosts | 必选 | (none) | String | 要连接到的一台或多台 Elasticsearch 主机,例如 ‘http://host_name:9092;http://host_name:9093’。 |
index | 必选 | (none) | String | Elasticsearch 中每条记录的索引。可以是一个静态索引(例如 ‘myIndex’)或一个动态索引(例如 'index-{log_ts |
document-type | 6.x 版本中必选 | (none) | String | Elasticsearch 文档类型。在 elasticsearch-7 中不再需要。 |
document-id.key-delimiter | 可选 | - | String | 复合键的分隔符(默认为"_“),例如,指定为” " 将导致文档 I D 为 " K E Y 1 "将导致文档 ID 为"KEY1 "将导致文档ID为"KEY1KEY2$KEY3"。 |
username | 可选 | (none) | String | 用于连接 Elasticsearch 实例的用户名。请注意,Elasticsearch 没有预绑定安全特性,但你可以通过如下指南启用它来保护 Elasticsearch 集群。 |
password | 可选 | (none) | String | 用于连接 Elasticsearch 实例的密码。如果配置了username,则此选项也必须配置为非空字符串。 |
failure-handler | 可选 | fail | String | 对 Elasticsearch 请求失败情况下的失败处理策略。有效策略为:fail:如果请求失败并因此导致作业失败,则抛出异常。ignore:忽略失败并放弃请求。retry-rejected:重新添加由于队列容量饱和而失败的请求。自定义类名称:使用 ActionRequestFailureHandler 的子类进行失败处理。 |
sink.delivery-guarantee | 可选 | AT_LEAST_ONCE | String | 承诺时可选择交付保证。有效值为:EXACTLY_ONCE:在故障转移情况下,记录也仅传递一次。AT_LEAST_ONCE:确保记录被传递,但可能会发生同一条记录被传递多次的情况。NONE:尽力提供记录。 |
sink.flush-on-checkpoint | 可选 | true | Boolean | 在进行 checkpoint 时是否保证刷出缓冲区中的数据。如果关闭这一选项,在进行checkpoint时 sink 将不再为所有进行 中的请求等待 Elasticsearch 的执行完成确认。因此,在这种情况下 sink 将不对至少一次的请求的一致性提供任何保证。 |
sink.bulk-flush.max-actions | 可选 | 1000 | Integer | 每个批量请求的最大缓冲操作数。 可以设置为’0’来禁用它。 |
sink.bulk-flush.max-size | 可选 | 2mb | MemorySize | 每个批量请求的缓冲操作在内存中的最大值。单位必须为 MB。 可以设置为’0’来禁用它。 |
sink.bulk-flush.interval | 可选 | 1s | Duration | flush 缓冲操作的间隔。 可以设置为’0’来禁用它。注意,'sink.bulk-flush.max-size’和’sink.bulk-flush.max-actions’都设置为’0’的这种 flush 间隔设置允许对缓冲操作进行完全异步处理。 |
sink.bulk-flush.backoff.strategy | 可选 | DISABLED | String | 指定在由于临时请求错误导致任何 flush 操作失败时如何执行重试。有效策略为:DISABLED:不执行重试,即第一次请求错误后失败。CONSTANT:等待重试之间的回退延迟。EXPONENTIAL:先等待回退延迟,然后在重试之间指数递增。 |
sink.bulk-flush.backoff.max-retries | 可选 | (none) | Integer | 最大回退重试次数。 |
sink.bulk-flush.backoff.delay | 可选 | (none) | Duration | 每次退避尝试之间的延迟。对于 CONSTANT 退避策略,该值是每次重试之间的延迟。对于 EXPONENTIAL 退避策略,该值是初始的延迟。 |
connection.path-prefix | 可选 | (none) | String | 添加到每个 REST 通信中的前缀字符串,例如,‘/v1’。 |
connection.request-timeout | 可选 | (none) | Duration | 从连接管理器请求连接的超时时间。超时时间必须大于或者等于 0,如果设置为 0 则是无限超时。 |
connection.timeout | 可选 | (none) | Duration | 建立请求的超时时间 。超时时间必须大于或者等于 0 ,如果设置为 0 则是无限超时。 |
socket.timeout | 可选 | (none) | Duration | 等待数据的 socket 的超时时间 (SO_TIMEOUT)。超时时间必须大于或者等于 0,如果设置为 0 则是无限超时。 |
format | 可选 | json | String | Elasticsearch 连接器支持指定格式。该格式必须生成一个有效的 json 文档。 默认使用内置的 ‘json’ 格式。 |
Elasticsearch 将文档存储在 JSON 字符串中。因此数据类型映射介于 Flink 数据类型和 JSON 数据类型之间。 Flink 为 Elasticsearch 连接器使用内置的 ‘json’ 格式。
下表列出了 Flink 中的数据类型与 JSON 中的数据类型的映射关系。
Flink SQL类型 | JSON类型 |
---|---|
CHAR/VARCHAR/STRING | string |
BOOLEAN | boolean |
BINARY/VARBINARY | string with encoding: base64 |
DECIMAL | number |
TINYINT | number |
SMALLINT | number |
INT | number |
BIGINT | number |
FLOAT | number |
DOUBLE | number |
DATE | string with format: date |
TIME | string with format: time |
TIMESTAMP | string with format: date-time |
TIMESTAMP_WITH_LOCAL_TIME_ZONE | string with format: date-time (with UTC time zone) |
INTERVAL | number |
ARRAY | array |
MAP / MULTISET | object |
ROW | object |
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。