# CSV Format
#### 功能描述
CSV Format 允许我们基于CSV schema 进行解析和生成CSV 数据。目前的CSV schema 是基于table schema 推导出来的。更多具体使用可参考开源社区文档：[CSV Format](https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/table/formats/csv/)。
#### 支持的Connector
- Kafka
- Upsert Kafka
- FileSystem
 
#### 参数说明
表1参数说明 
| 参数                          | 是否必选                                   | 默认值    | 类型                                           | 说明                                                                                                                                            |
|:---|:---|:---|:---|:---|
| format                      | 是                                      | (none) | String                                       | 指定要使用的格式，这里应该是 'csv'。                                                                                                                         |
| csv.field-delimiter         | 否                                      | ，      | String                                       | 字段分隔符 (默认',')，必须为单字符。您可以使用反斜杠字符指定一些特殊字符，例如 '\\t' 代表制表符。 您也可以通过 unicode 编码在纯 SQL 文本中指定一些特殊字符，例如'csv.field-delimiter' = U\&'\\0001' 代表 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 值。                                                                                                                     |
   
#### 数据类型映射
目前 CSV 的 schema 都是从 table schema 推断而来的。显式地定义 CSV schema 暂不支持。 Flink 的 CSV Format 数据使用 [jackson databind API](https://github.com/FasterXML/jackson-databind) 去解析 CSV 字符串。
表2数据类型映射 
| Flink SQL 类型            | CSV 类型                        |
|:---|:---|
| 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 |
| INTERVAL                | number                        |
| ARRAY                   | array                         |
| ROW                     | object                        |
   
#### 示例
使用kafka发送数据，输出到print中。
1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列 \> 找到作业的所属队列 \> 更多 \> 测试地址连通性 \> 输入kafka的地址 \> 测试）。如果能连通，则表示跨源已经绑定成功；否则表示未成功。
2. 创建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' = 'kafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'csv'
   );
   CREATE TABLE printSink (
     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' = 'print'
   );
   insert into printSink select * from kafkaSource;
   ```
   
   
3. 向kafka的作为source的topic中插入下列数据： 
   ```
   202103251505050001,appShop,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
   ```
   
   
4. 按照如下方式查看taskmanager.out文件中的数据结果： 
   1. 登录DLI管理控制台，选择"作业管理 \> Flink作业"。
   
   2. 单击对应的Flink作业名称，选择"运行日志"，单击"OBS桶"，根据作业运行的日期，找到对应日志的文件夹。
   
   3. 进入对应日期的文件夹后，找到名字中包含"taskmanager"的文件夹进入，下载获取.out文件查看结果日志。
   
   
   ```
   +I[202103251505050001, appShop, 2021-03-25 15:05:05, 500.0, 400.0, 2021-03-25 15:10:00, 0003, Cindy, 330108]
   +I[202103241606060001, appShop, 2021-03-24 16:06:06, 200.0, 180.0, 2021-03-24 16:10:06, 0001, Alice, 330106]
   ```