读CSV文件

功能介绍

读CSV文件。支持从本地、hdfs、http读取

参数说明

名称 中文名称 描述 类型 是否必须? 默认值
filePath 文件路径 文件路径 String
schemaStr Schema Schema。格式为”colname coltype[, colname2, coltype2[, …]]”,例如”f0 string, f1 bigint, f2 double” String
fieldDelimiter 字段分隔符 字段分隔符 String “,”
quoteChar 引号字符 引号字符 Character “\””
skipBlankLine 是否忽略空行 是否忽略空行 Boolean true
rowDelimiter 行分隔符 行分隔符 String “\n”
ignoreFirstLine 是否忽略第一行数据 是否忽略第一行数据 Boolean false

支持的字段类型包括:

字段类型 描述 值域 Flink类型 Java类型
VARCHAR/STRING 可变长度字符串 最大容量为4mb Types.STRING java.lang.String
BOOLEAN 逻辑值 值:TRUE,FALSE,UNKNOWN Types.BOOLEAN java.lang.Boolean
TINYINT 微整型,1字节整数 范围是-128到127 Types.BYTE java.lang.Byte
SMALLINT 短整型,2字节整数 范围为-32768至32767 Types.SHORT java.lang.Short
INT 整型,4字节整数 范围是-2147483648到2147483647 Types.INT java.lang.Integer
BIGINT/LONG 长整型,8字节整数 范围是-9223372036854775808至9223372036854775807 Types.LONG java.lang.Long
FLOAT 4字节浮点数 6位数字精度 Types.FLOAT java.lang.Float
DOUBLE 8字节浮点数 15位十进制精度 Types.DOUBLE java.lang.Double
DECIMAL 小数类型 示例:123.45是DECIMAL(5,2)值 Types.DECIMAL java.math.BigDecimal
DATE 日期 示例:’1969-07-20’ Types.SQL_DATE java.sql.Date
TIME 时间 示例:’20:17:40’ Types.SQL_TIME java.sql.Time
TIMESTAMP 时间戳,日期和时间 示例:’1969-07-20 20:17:40’ Types.SQL_TIMESTAMP java.sql.Timestamp

关于分隔符的说明:

Web前端支持用户输入如下转义字符和unicode字符作为分隔符:

输入分隔符 含义
\t 制表符(tab键)
\n 换行符
\b 退格符
\r 回车
\f 换页
\\ 反斜线字符
单引号
双引号
\ddd 1到3位八进制数所代表的任意字符,例如\001表示’ctrl + A’, \40表示空格符
\udddd 1到4位十六进制数所代表unicode字符,例如\u0001表示’ctrl + A’, \u0020表示空格符

脚本示例

Csv Batch Source

  1. filePath = 'http://alink-dataset.cn-hangzhou.oss.aliyun-inc.com/csv/iris.csv'
  2. schema = 'sepal_length double, sepal_width double, petal_length double, petal_width double, category string'
  3. csvSource = CsvSourceBatchOp()\
  4. .setFilePath(filePath)\
  5. .setSchemaStr(schema)\
  6. .setFieldDelimiter(",")
  7. BatchOperator.collectToDataframe(csvSource)

脚本运行结果

  1. sepal_length sepal_width petal_length petal_width category
  2. 0 6.3 3.3 6.0 2.5 Iris-virginica
  3. 1 5.6 2.8 4.9 2.0 Iris-virginica
  4. 2 5.0 3.3 1.4 0.2 Iris-setosa
  5. 3 5.8 2.7 5.1 1.9 Iris-virginica
  6. 4 7.0 3.2 4.7 1.4 Iris-setosa

Csv Stream Source

  1. filePath = 'http://alink-dataset.cn-hangzhou.oss.aliyun-inc.com/csv/iris.csv'
  2. schema = 'sepal_length double, sepal_width double, petal_length double, petal_width double, category string'
  3. csvSource = CsvSourceStreamOp()\
  4. .setFilePath(filePath)\
  5. .setSchemaStr(schema)\
  6. .setFieldDelimiter(",")
  7. csvSource.print()
  8. StreamOperator.execute()

脚本运行结果

  1. sepal_length sepal_width petal_length petal_width category
  2. 1 5.5 2.4 3.8 1.1 Iris-versicolor
  3. 2 6.1 2.6 5.6 1.4 Iris-virginica
  4. 3 6.0 2.2 4.0 1.0 Iris-versicolor
  5. 4 5.5 2.4 3.7 1.0 Iris-versicolor
  6. 5 4.6 3.1 1.5 0.2 Iris-setosa