Flink Version | Description |
1.13 | Supported |
1.14 | Not supported. |
1.16 | Not supported. |
CREATE TABLE `rmq_sink_json_table` (`id` int,`name` STRING) WITH ('connector' = 'rabbitmq', -- Must be 'rabbitmq''host' = 'xxxx', -- rabbitmq host'port' = 'xxxx', -- rabbitmq port'vhost' = '/', -- Virtual host'username' = 'xxxx', -- Username'password' = 'xxxx', -- User password'exchange' = 'exchange', -- Exchange name'routing-Key' = 'Key', -- Binding Key'format' = 'json', -- Define the data format (JSON).'json.fail-on-missing-field' = 'false', -- If set to false, no error will be reported when encountering missing fields.'json.ignore-parse-errors' = 'true' -- If set to true, any parse errors will be ignored.);
CREATE TABLE `rmq_sink_csv_table` (`id` int,`name` STRING) WITH ('connector' = 'rabbitmq', -- Must be 'rabbitmq''host' = 'xxxx', -- rabbitmq host'port' = 'xxxx', -- rabbitmq port'vhost' = '/', -- Virtual host'username' = 'xxxx', -- Username'password' = 'xxxx', -- User password'exchange' = 'exchange', -- Exchange name'routing-Key' = 'Key', -- Binding Key'format' = 'csv' -- Define the data format (CSV).);
Parameter Value | Required | Default Value | Description |
connector | Yes | N/A | Must be set to 'rabbitmq'. |
host | Yes | N/A | The host where the queue resides. |
port | Yes | 5672 | rabbitmq port. |
vhost | Yes | / | Virtual host. |
username | Yes | guest | Role name. |
password | Yes | guest | Role password. |
queue | No | N/A | Queue name. |
exchange | Yes | N/A | Exchange name. |
routing-key | No | N/A | Exchange binding Key. |
delivery-mode | No | 1 | Whether the message is persistent. 1: Non-persistent. 2: Persistent. |
expiration | No | 86400000 | The message expiration time is one day by default (unit: milliseconds). |
network-recovery-interval | No | 30s | Network recovery interval. |
automatic-recovery | No | true | rabbitmq is automatically connected by default. |
topology-recovery | No | true | rabbitmq topology is automatically recovered by default. |
connection-timeout | No | 30s | The connection timeout is 30s by default. |
requested-frame-max | No | 0 | The maximum communication frame size for the initial request, in bytes. A value of 0 indicates no limit. |
requested-heartbeat | No | 60s | The request heartbeat times out. |
prefetch-count | No | 0 | The maximum number of messages sent by the server. A value of 0 indicates no limit. (Supported only in version 1.13.) |
delivery-timeout | No | 30s | The submission queue timeout. (Supported only in version 1.13.) |
format | Yes | N/A | The input/output format for RMQ messages. Currently supported formats include 'csv' and 'json'. |
Parameter Value | Required | Default Value | Description |
json.fail-on-missing-field | No | false | If this is true, the job will fail in case of missing parameters. If this is false (default), the missing parameters will be set to null and the job will continue to be executed. |
json.ignore-parse-errors | No | false | If this is true, when there is a parse error, the field will be set to null and the job will continue to be executed. If this is false, the job will fail in case of a parse error. |
json.timestamp-format.standard | No | SQL | Specify the format for JSON timestamp fields. The default is SQL (format: yyyy-MM-dd HH:mm:ss.s{optional precision}). You can also choose ISO-8601, with the format yyyy-MM-ddTHH:mm:ss.s{optional precision}. |
Parameter Value | Required | Default Value | Description |
csv.field-delimiter | No | , | The CSV field delimiter, which is a half-width comma by default. |
csv.line-delimiter | No | U&'\\000A' | Specify the CSV row delimiter. The default is the newline character \\n, which must be represented as U&'\\000A' in SQL. If you need to use the carriage return character \\r, it must be represented as U&'\\000D' in SQL. (Supported only in version 1.11.) |
csv.disable-quote-character | No | false | Whether to disable quote characters. If this is true, 'csv.quote-character' cannot be used. |
csv.quote-character | No | '' | The quote character for field enclosure. Content inside quotes is treated as a whole. The default is ''. |
csv.ignore-parse-errors | No | false | Whether to ignore parse errors. If this is true, fields will be set to null in case of parse failure. |
csv.allow-comments | No | false | Whether to ignore comment lines that start with # and output them as empty lines (if this is true, make sure you set csv.ignore-parse-errors to true as well). |
csv.array-element-delimiter | No | ; | The delimiter for array elements. The default is ;. |
csv.escape-character | No | N/A | The escape character. By default, escape characters are disabled. |
csv.null-literal | No | N/A | The string that will be seen as null. |
-- Note: Replace the parameters with the information of the cluster to which they belong.CREATE TABLE `rabbitmq_source_json_table` (`id` INT, `name` STRING) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://host:port/database?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai','table-name' = 'source_table_name','username' = 'username','password' = 'password');CREATE TABLE `rabbitmq_sink_json_table` (`id` INT, `name` STRING) WITH ('connector' = 'rabbitmq','host' = 'host','port' = 'port','vhost' = 'vhost','username' = 'username','password' = 'password','queue' = 'queue-name','exchange'='exchange','routing-key'='key','format' = 'json');insert into rabbitmq_sink_json_table select * from rabbitmq_source_json_table;
Esta página foi útil?
Você também pode entrar em contato com a Equipe de vendas ou Enviar um tíquete em caso de ajuda.
comentários