Connectors如何讀寫csv文件，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

阜城ssl適用于網(wǎng)站、小程序/APP、API接口等需要進行數(shù)據(jù)傳輸應用場景，ssl證書未來市場廣闊！成為創(chuàng)新互聯(lián)公司的ssl證書銷售渠道，可以享受市場價格4-6折優(yōu)惠！如果有意向歡迎電話聯(lián)系或者加微信：18982081108（備注：SSL證書合作）期待與您的合作！

通過使用Flink DataSet Connectors 數(shù)據(jù)流連接器打開csv文件，并提供數(shù)據(jù)流輸入與輸出操作；

示例環(huán)境

java.version: 1.8.xflink.version: 1.11.1

示例數(shù)據(jù)源 （項目碼云下載）

Flink 系例 之 搭建開發(fā)環(huán)境與數(shù)據(jù)

示例模塊 （pom.xml）

Flink 系例 之 DataStream Connectors 與 示例模塊

數(shù)據(jù)流輸入

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> dataSet = env
                .readCsvFile(filePath)
                .fieldDelimiter(",")
                .types(Integer.class, String.class, Integer.class, Integer.class, String.class, String.class, Long.class);
        //打印
        dataSet.print();
    }

}

數(shù)據(jù)流輸出

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數(shù)據(jù)寫入到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";
        //添加數(shù)據(jù)
        Tuple7 row = new Tuple7<>(15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L);
        //轉換為dataSet
        DataSet> dataSet = env.fromElements(row);
        //將內容寫入到File中，如果文件已存在，將會被復蓋
        dataSet.writeAsCsv(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        env.execute("fline file sink");
    }

}

數(shù)據(jù)展示

看完上述內容，你們掌握Connectors如何讀寫csv文件的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝各位的閱讀！