TFRecordDataset文件导出 (TFRecordDatasetSinkStreamOp)

Java 类名：com.alibaba.alink.operator.stream.sink.TFRecordDatasetSinkStreamOp
Python 类名：TFRecordDatasetSinkStreamOp

功能介绍

写出 TFRecordDataset 文件（TFRecordDataset 的介绍可以参考 TensorFlow 文档：https://www.tensorflow.org/tutorials/load_data/tfrecord ）。

使用说明

需要指定文件路径 filePath，默认为单并行度写出单个文件。
如果希望并行的写出文件，那么需要设置参数numFiles，得到的是一个包含多个 TFRecordDataset 文件的目录。
TFRecord 中， Feature 允许的数据类型仅有 float, int64, bytes。
其中数据类型为 bytes时，存储的数据实际上为 ByteString 的列表；为 float, int64 时存储的数据为 float 或 int64 的列表。
数据写出时会根据在 Alink 中的类型进行转换， 其他类型请先使用类型转换组件进行转换：

• DOUBLE, FLOAT, BIG_DEC：转为float特征；
• LONG, INT, BIG_INT, SHORT：转为int64特征；
• STRING：转为bytes特征，按 UTF8 编码对应 1 个ByteString；
• DENSE_VECTOR：转为float特征；
• FLOAT_TENSOR, DOUBLE_TENSOR：转为float特征，数据被展平为1维；
• INT_TENSOR, LONG_TENSOR：转为int64特征，数据被展平为1维；
• BYTE_TENSOR：转为bytes特征，rank = 1 时对应 1 个ByteString，rank = 2时对应ByteString的列表，其他 rank 不支持；
• STRING_TENSOR：转为bytes特征，按 UTF8 编码对应ByteString的列表；
• VARBINARY：转为bytes特征，对应 1 个ByteString。

  参数说明

  | 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| filePath | 文件路径 | 文件路径 | String | ✓ | | |

| numFiles | 文件数目 | 文件数目 | Integer | | | 1 |

| overwriteSink | 是否覆写已有数据 | 是否覆写已有数据 | Boolean | | | false |

代码示例

Python 代码

schemaStr = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string"
source = CsvSourceStreamOp() \
    .setFilePath("https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv") \
    .setSchemaStr(schemaStr)
sink = TFRecordDatasetSinkStreamOp() \
    .setFilePath("/tmp/iris.tfrecord") \
    .setOverwriteSink(True) \
    .linkFrom(source)
StreamOperator.execute()

Java 代码

import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.sink.TFRecordDatasetSinkStreamOp;
import com.alibaba.alink.operator.stream.source.CsvSourceStreamOp;
import org.junit.Test;
public class TFRecordDatasetSinkStreamOpTest {
    @Test
    public void testTFRecordDatasetSinkStreamOp() throws Exception {
        String schemaStr
            = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";
        StreamOperator <?> source = new CsvSourceStreamOp()
            .setFilePath("https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv")
            .setSchemaStr(schemaStr);
        StreamOperator <?> sink = new TFRecordDatasetSinkStreamOp()
            .setFilePath("/tmp/iris.tfrecord")
            .setOverwriteSink(true)
            .linkFrom(source);
        StreamOperator.execute();
    }
}