Alink 是基于 Flink 的机器学习算法平台，欢迎访问 Alink 的 GitHub 获取更多信息。本文主要分享 Alink 的使用技巧之一，如何使用批式 CSV 进行数据读取。

基本操作

我们先下载个 csv 文件用作后面的测试数据。

将数据文件下载到本地，文件路径为 /Users/yangxu/flinkml/data/iris/iris.data，使用文本编辑器打开如下所示，每行为一条数据，每条数据包括4个数值字段和一个字符串字段，各字段间使用逗号分隔。

数据下载
http://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data

读取本地数据

使用 CsvSourceBatchOp 可以批式读 CSV 格式文件，其必填的两个参数为：filePath 和 schemaStr。filePath 为 CSV 格式文件所在的路径；schemaStr 为数据各字段的名称和类型。关于 Schema String 更多的介绍可以参见：

Alink Schema String简介
https://zhuanlan.zhihu.com/p/103027854

可以使用如下脚本，读取数据，并取前5条数据打印显示出来。

source_local = CsvSourceBatchOp()\
.setFilePath("/Users/yangxu/flinkml/data/iris/iris.data")\
.setSchemaStr("sepal_length double, sepal_width double, petal_length double, petal_width double, category string")
source_local.firstN(5).print()

数据打印显示如下：

注意：CsvSourceBatchOp 必填的参数 filePath 和 schema 进行赋值，filePath 为本地文件存储路径"/Users/yangxu/flinkml/data/iris/iris.data"，schema 为 iris 数据集的列名和类型信息，共有5个字段：sepal_length, sepal_width, petal_length, petal_width, category；其数据类型分别为 double, double, double, double, string。

读取网络数据

相对于前面介绍的读取本地 CSV 数据，我们只需将数据存储路径参数 filePath，赋值为 http 路径地址即可。脚本如下：

source_url = CsvSourceBatchOp()\
.setFilePath("http://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data")\
.setSchemaStr("sepal_length double, sepal_width double, petal_length double, petal_width double, category string")
source_url.firstN(5).print()

数据打印显示如下：

参数详细说明

Alink CsvSourceBatchOp 提供了丰富的读取功能，具体内容可以参阅其参数说明文档：

https://github.com/alibaba/Alink/blob/master/docs/cn/csvsourcebatchop.md

更复杂的例子

对于葡萄酒品质数据集，我们将其下载达到本地，可以看到其文件内容。

数据下载
http://archive.ics.uci.edu/ml/machine-learning-databases/wine-quality/winequality-white.csv

第一行为数据列名的说明，第二行开始是数据，可以看到都是数值类型，各个数之间用分号“;”进行分隔。

我们可以通过设置参数 ignoreFirstLine 为 True，略过第一行；并且可以设置字段分隔符参数 fieldDelimiter 为分号“;”。另外，由于列名不能包含空格，所以由文件第一列转化来的列名需要进行相应处理，这里我们将其写为驼峰形式；并加上数据类型，这里都是 double 类型，构成了数据集的 SchemaStr。具体的脚本如下：

wine_url = CsvSourceBatchOp()\
.setFilePath("http://archive.ics.uci.edu/ml/machine-learning-databases/wine-quality/winequality-white.csv")\
.setSchemaStr("fixedAcidity double,volatileAcidity double,citricAcid double,residualSugar double,"+\
"chlorides double,freeSulfurDioxide double,totalSulfurDioxide double,density double,"+\
"pH double,sulphates double,alcohol double,quality double")\
.setFieldDelimiter(";")\
.setIgnoreFirstLine(True);
wine_url.firstN(5).print()

数据打印显示如下：

可以看到由于每行的数据较多，在显示的时候进行了分段显示，每段数据可以由其最左方的行索引编号关联。

扫码关注「流计算Alink」，获取更多机器学习干货