Connectors如何读写csv文件

Connectors如何读写csv文件，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。

阜城ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为创新互联公司的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：18982081108（备注：SSL证书合作）期待与您的合作！

通过使用Flink DataSet Connectors 数据流连接器打开csv文件，并提供数据流输入与输出操作；

示例环境

java.version: 1.8.xflink.version: 1.11.1

示例数据源 （项目码云下载）

Flink 系例 之 搭建开发环境与数据

示例模块 （pom.xml）

Flink 系例 之 DataStream Connectors 与 示例模块

数据流输入

CsvSource.java

package com.flink.examples.file;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple7;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description 从csv文件中读取内容输出到DataSet中
 */
public class CsvSource {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String filePath = "D:\\Workspaces\\idea_2\\flink-examples\\connectors\\src\\main\\resources\\user.csv";
        DataSet<Tuple7<Integer, String, Integer, Integer, String, String, Long>> dataSet = env
                .readCsvFile(filePath)
                .fieldDelimiter(",")
                .types(Integer.class, String.class, Integer.class, Integer.class, String.class, String.class, Long.class);
        //打印
        dataSet.print();
    }

}

数据流输出

CsvSink.java

package com.flink.examples.file;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple7;
import org.apache.flink.core.fs.FileSystem;

/**
 * @Description 将DataSet数据写入到csv文件中
 */
public class CsvSink {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //需先建立文件
        String filePath = "D:\\Workspaces\\idea_2\\flink-examples\\connectors\\src\\main\\resources\\user.csv";
        //添加数据
        Tuple7<Integer, String, Integer, Integer, String, String, Long> row = new Tuple7<>(15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L);
        //转换为dataSet
        DataSet<Tuple7<Integer, String, Integer, Integer, String, String, Long>> dataSet = env.fromElements(row);
        //将内容写入到File中，如果文件已存在，将会被复盖
        dataSet.writeAsCsv(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        env.execute("fline file sink");
    }

}

数据展示

看完上述内容，你们掌握Connectors如何读写csv文件的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注创新互联行业资讯频道，感谢各位的阅读！

文章名称：Connectors如何读写csv文件
网页网址：https://www.cdcxhl.com/article34/jopspe.html

成都网站建设公司_创新互联，为您提供品牌网站建设、手机网站建设、搜索引擎优化、品牌网站设计、网站内链、标签优化

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联