文档首页/ 数据湖探索 DLI/ Flink SQL语法参考/ Flink Opensource SQL1.15语法参考/ Format/ CSV Format
更新时间：2024-04-19 GMT+08:00
查看PDF
分享

CSV Format

功能描述

CSV Format 允许我们基于CSV schema 进行解析和生成CSV 数据。目前的CSV schema 是基于table schema 推导出来的。更多具体使用可参考开源社区文档：CSV Format。

支持的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 去解析 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]

父主题： Format