CSV Format
功能描述
CSV Format 允许我们基于CSV schema 进行解析和生成CSV 数据。目前的CSV schema 是基于table schema 推导出来的。
支持的Connector
- Kafka
- Upsert Kafka
参数说明
|
参数 |
是否必选 |
默认值 |
类型 |
说明 |
|---|---|---|---|---|
|
format |
是 |
(none) |
String |
指定要使用的格式，这里应该是 'csv'。 |
|
csv.field-delimiter |
否 |
， |
String |
字段分隔符 (默认',')，必须为单字符。您可以使用反斜杠字符指定一些特殊字符，例如 '\t' 代表制表符。 您也可以通过 unicode 编码在纯 SQL 文本中指定一些特殊字符，例如 'csv.field-delimiter' = '\u0001' 代表 0x01 字符。 |
|
csv.disable-quote-character |
否 |
false |
Boolean |
是否禁止对引用的值使用引号 (默认是 false). 如果禁止，选项 'csv.quote-character' 不能设置。 |
|
csv.quote-character |
否 |
‘’ |
String |
用于围住字段值的引号字符 (默认"). |
|
csv.allow-comments |
否 |
false |
Boolean |
是否允许忽略注释行（默认不允许），注释行以 '#' 作为起始字符。 如果允许注释行，请确保 csv.ignore-parse-errors 也开启了从而允许空行。 |
|
csv.ignore-parse-errors |
否 |
false |
Boolean |
当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。 |
|
csv.array-element-delimiter |
否 |
； |
String |
分隔数组和行元素的字符串(默认';'). |
|
csv.escape-character |
否 |
(none) |
String |
转义字符(默认关闭). |
|
csv.null-literal |
否 |
(none) |
String |
是否将 "null" 字符串转化为 null 值。 |
示例
使用kafka发送数据，输出到print中。
- 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列-->找到作业的所属队列-->更多-->测试地址连通性-->输入kafka的地址-->测试）。若能连通，则表示跨源已经绑定成功；否则表示未成功。
- 创建flink opensource sql作业，并提交运行，其代码如下：
CREATE TABLE kafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = '<yourSourceTopic>', 'properties.bootstrap.servers' = '<yourKafkaAddress>:<yourKafkaPort>', 'properties.group.id' = '<yourGroupId>', 'scan.startup.mode' = 'latest-offset', "format" = "csv" ); CREATE TABLE kafkaSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = '<yourSinkTopic>', 'properties.bootstrap.servers' = '<yourKafkaAddress>:<yourKafkaPort>', "format" = "csv" ); insert into kafkaSink select * from kafkaSource;
- 向kafka的作为source的topic中插入下列数据：
202103251505050001,qqShop,2021-03-25 15:05:05,500.00,400.00,2021-03-25 15:10:00,0003,Cindy,330108 202103241606060001,appShop,2021-03-24 16:06:06,200.00,180.00,2021-03-24 16:10:06,0001,Alice,330106
- 读取kafka中作为sink的topic，结果如下：
202103251505050001,qqShop,"2021-03-25 15:05:05",500.0,400.0,"2021-03-25 15:10:00",0003,Cindy,330108 202103241606060001,appShop,"2021-03-24 16:06:06",200.0,180.0,"2021-03-24 16:10:06",0001,Alice,330106
