本篇內(nèi)容主要講解“如何用Flink Connectors讀寫txt文件”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“如何用Flink Connectors讀寫txt文件”吧!
成都創(chuàng)新互聯(lián)-專業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設(shè)、高性價(jià)比旌德網(wǎng)站開發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫,直接使用。一站式旌德網(wǎng)站制作公司更省心,省錢,快速模板網(wǎng)站建設(shè)找我們,業(yè)務(wù)覆蓋旌德地區(qū)。費(fèi)用合理售后完善,十余年實(shí)體公司更值得信賴。
通過使用Flink DataSet Connectors 數(shù)據(jù)流連接器打開txt文件,并提供數(shù)據(jù)流輸入與輸出操作;
示例環(huán)境
java.version: 1.8.xflink.version: 1.11.1
示例數(shù)據(jù)源 (項(xiàng)目碼云下載)
Flink 系例 之 搭建開發(fā)環(huán)境與數(shù)據(jù)
示例模塊 (pom.xml)
Flink 系例 之 DataStream Connectors 與 示例模塊
數(shù)據(jù)流輸入
TextSource.java
package com.flink.examples.file; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.time.DateUtils; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.util.Collector; import scala.Tuple7; import java.util.Date; /** * @Description 從txt文件中讀取內(nèi)容輸出到DataSet中 */ public class TextSource { 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.txt"; DataSet<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> dataSet = env.readTextFile(filePath) .flatMap(new FlatMapFunction<String, Tuple7<Integer, String, Integer, Integer, String, Date, Long>>() { @Override public void flatMap(String value, Collector<Tuple7<Integer, String, Integer, Integer, String, Date, Long>> out) throws Exception { if (StringUtils.isNotBlank(value)) { String[] values = value.split(","); out.collect(Tuple7.apply( Integer.parseInt(values[0]), values[1], Integer.parseInt(values[2]), Integer.parseInt(values[3]), values[4], DateUtils.parseDate(values[5], "yyyy-MM-dd HH:mm:ss"), Long.parseLong(values[6]))); } } }); dataSet.print(); } }
數(shù)據(jù)流輸出
TextSink.java
package com.flink.examples.file; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.core.fs.FileSystem; import org.apache.flink.types.Row; /** * @Description 將DataSet數(shù)據(jù)寫入到txt文件中 */ public class TextSink { 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.txt"; //將實(shí)體轉(zhuǎn)換為Row對象,new Row(字段個(gè)數(shù)) Row row = Row.of(15, "chen1", 40, 1, "CN", "2020-09-08 00:00:00", 1599494400000L); //轉(zhuǎn)換為dataSet DataSet<Row> dataSet = env.fromElements(row); //將內(nèi)容寫入到File中,如果文件已存在,將會(huì)被復(fù)蓋 dataSet.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(1); env.execute("fline file sink"); } }
數(shù)據(jù)展示
到此,相信大家對“如何用Flink Connectors讀寫txt文件”有了更深的了解,不妨來實(shí)際操作一番吧!這里是創(chuàng)新互聯(lián)網(wǎng)站,更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢,關(guān)注我們,繼續(xù)學(xué)習(xí)!
新聞名稱:如何用FlinkConnectors讀寫txt文件
轉(zhuǎn)載來于:http://bm7419.com/article24/pscece.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供自適應(yīng)網(wǎng)站、網(wǎng)站設(shè)計(jì)、品牌網(wǎng)站建設(shè)、網(wǎng)站收錄、網(wǎng)站改版、營銷型網(wǎng)站建設(shè)
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)