2. 程式人生 > >Apache Flink 各類關鍵資料格式讀取/SQL支援

Apache Flink 各類關鍵資料格式讀取/SQL支援

阿新 • • 發佈：2018-12-25

目前事件歸併分為兩種，一種為實時的歸併，即基於Kafka內的資料進行歸併和事件生成；一種是週期性的歸併，即基於Hive中的資料進行資料的歸併和事件生成。

基於SQL歸併時Spark Streaming支援的輸入/輸出資料如下：

資料型別

Flink支援情況
Kafka 需要定義schema
HDFS(parquet/csv/textfile)

讀取parquet需要使用AvroParquetInputFormat

csv/textfile有readCsvFile和TextFileInput
Hive

1.需要啟用hive service metastore來提供thrift metastore介面

2.需要依賴flink-hcatalog來進行讀取
JDBC(PostgreSQL） JDBCInputFormat

下面就Apache Flink是否支援上述格式進行測試。

1.Kafka

首先需要定義一個POJO類，用於代表從kafka讀取的dstream裡的內容：

package com.flinklearn.models;
 
/**
 * Created by dell on 2018/10/23.
 */
public class TamAlert {
    private String msg;
 
    public TamAlert(){}
 
    public String getMsg() {
        return msg;
    }
 
    public void setMsg(String msg) {
        this.msg = msg;
    }
}

其次，在Flink的DataStream上執行SQL與Spark比較不同，對於Spark而言一直是stream的transform、registerTempTable動作，而在Flink上需要將DataStream轉換為Table，才能執行相關SQL，而如果要進行transform需要再次將Table轉為DataFrame才可以。

程式碼如下：

package com.flinklearn.main
 
import java.util.Properties
 
import com.alibaba.fastjson.{JSON}
import com.flinklearn.models.TamAlert
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{TableEnvironment, Types}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
 
import scala.collection.mutable.ArrayBuffer
 
/**
 * Created by dell on 2018/10/22.
 */
class Main {
  def startApp(): Unit = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "brokerserver")
    properties.setProperty("group.id", "com.flinklearn.main.Main")
    val env = StreamExecutionEnvironment.getExecutionEnvironment
 
    //從kafka讀取資料，得到stream
    val stream:DataStream[TamAlert] = env
      .addSource(new FlinkKafkaConsumer010[String]("mytopic", new SimpleStringSchema(), properties))
      .map(line => {
        var rtn:TamAlert = null
        try{
          val temp = JSON.parseObject(line).getJSONObject("payload")
          rtn = new TamAlert()
          rtn.setMsg(temp.getString("msg"))
        }catch{
          case ex:Exception => {
            ex.printStackTrace()
          }
        }
        rtn
      }).filter(line=>line!=null)
 
    //將stream註冊為temp_alert表，並列印msg欄位
    val tableEnv:StreamTableEnvironment = TableEnvironment.getTableEnvironment(env)
    tableEnv.registerDataStream("temp_alert", stream,
      'msg)
    val httpTable  = tableEnv.sqlQuery("select msg from temp_alert")
 
    val httpStream = tableEnv.toAppendStream[(String,String,Integer)](httpTable)
    httpStream.print()
 
    env.execute("Kafka sql test.")
  }
}
object Main {
  def main(args:Array[String]):Unit = {
    new Main().startApp()
  }
}

2.HDFS

2.1 Parquet

對於HDFS Parquet格式的資料，Apache Flink並不如Spark一般有十分方便的read.parquet()介面，需要藉助AvroParquetInputFormat來讀取相應檔案。具體操作步驟如下：

1.在pom.xml中引入相應的依賴

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-hadoop-compatibility_2.11</artifactId>
  <version>1.6.1</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-avro</artifactId>
  <version>1.6.1</version>
</dependency>
<dependency>
  <groupId>org.apache.parquet</groupId>
  <artifactId>parquet-avro</artifactId>
  <version>1.10.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-mapreduce-client-core</artifactId>
  <version>3.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-hdfs</artifactId>
  <version>3.1.1</version>
</dependency>
<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-core</artifactId>
  <version>1.2.1</version>
</dependency>

2.使用avsc檔案定義schema

{"namespace": "com.flinklearn.models",
 "type": "record",
 "name": "AvroTamAlert",
 "fields": [
    {"name": "msg", "type": ["string","null"]}
 ]
}

3.使用avro-tools生成對應的java類，並將java檔案拷貝到專案裡，本例子中是AvroTamAlert.java：

4.使用AvroParquetInputFormat來讀取parquet檔案：

package com.flinklearn.main
 
import java.util.Arrays
 
import com.flinklearn.models.{AvroTamAlert}
import org.apache.avro.Schema
import org.apache.avro.util.Utf8
import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.api.scala.{ExecutionEnvironment}
import org.apache.flink.table.api.TableEnvironment
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat}
import org.apache.flink.api.scala._
import org.apache.parquet.avro.AvroParquetInputFormat
import org.apache.flink.table.api.scala._
 
/**
 * Created by dell on 2018/10/23.
 */
class Main {
  def startApp(): Unit ={
    val env = ExecutionEnvironment.getExecutionEnvironment
 
    val job = Job.getInstance()
    val dIf = new HadoopInputFormat[Void, AvroTamAlert](new AvroParquetInputFormat(), classOf[Void], classOf[AvroTamAlert], job)
    FileInputFormat.addInputPath(job, new Path("/user/hive/warehouse/xx.db/yy"))
 
    AvroParquetInputFormat.setAvroReadSchema(job, AvroTamAlert.getClassSchema)
 
    val dataset = env.createInput(dIf).map(line=>line.f1).map(line=>(line.getSip.toString,line.getDip.toString,line.getDport))
 
    val tableEnv = TableEnvironment.getTableEnvironment(env)
    tableEnv.registerDataSet("tmp_table", dataset, 'msg)
 
    val table = tableEnv.sqlQuery("select msg from tmp_table")
    tableEnv.toDataSet[(String,String,Integer)](table).print()
 
    env.execute("start hdfs parquet test")
  }
}
object Main {
  def main(args:Array[String]):Unit = {
    new Main().startApp()
  }
}

2.2 CSV

需要新增的引數在2.3小節中。

package com.flinklearn.main
 
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.TableEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
/**
 * Created by dell on 2018/10/25.
 */
object Main {
  def main(args:Array[String]):Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val dataset:DataSet[(String,Integer)] = env.readCsvFile("hdfs://ip:8020/mytest")
 
    val tableEnv = TableEnvironment.getTableEnvironment(env)
    tableEnv.registerDataSet("tmp_table", dataset, 'name, 'num)
    val table = tableEnv.sqlQuery("select name,num from tmp_table")
 
    val rtnDataset = tableEnv.toDataSet[(String,Integer)](table)
    rtnDataset.print()
 
    env.execute("test hdfs csvfile")
  }
}

2.3 TextFile

有幾個關鍵的引數必須加到flink-conf.yaml檔案中：

第一個引數指定Hadoop的配置檔案

第二個引數指定模式為舊模式，因為flink1.6.1是用的scala2.11，使用scala介面會存在一定的問題（報jobgraph生成失敗，目前還不清楚具體原因）

第三個引數指定類載入順序（如果不指定，會報hdfs 不可讀取塊錯誤）

同時，需要將flink-hadoop-compatibility_2.11-1.6.1.jar放到flink/lib資料夾下，pom裡打包沒有用。

以上操作做完，就可以正確的讀取hdfs上的檔案了：

package com.flinklearn.main
 
import org.apache.flink.api.java.io.RowCsvInputFormat
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.hadoop.mapreduce.HadoopInputFormat
import org.apache.flink.hadoopcompatibility.scala.HadoopInputs
import org.apache.flink.table.api.TableEnvironment
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.{CombineTextInputFormat, TextInputFormat, FileInputFormat}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
/**
 * Created by dell on 2018/10/25.
 */
object Main {
  def main(args:Array[String]):Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val dataset:DataSet[(LongWritable,Text)] = env.createInput(HadoopInputs.readHadoopFile[LongWritable,Text](
      new CombineTextInputFormat,
      classOf[LongWritable],
      classOf[Text],
      "/mytest"
    ))
    val transDataset = dataset.map(line=>{
      val lines = line._2.toString.split(",")
      if(lines.length == 2){
        (lines(0).toString,lines(1).toInt)
      }else{
        null
      }
    }).filter(line=>line!=null)
 
    print(transDataset.count())
 
    val tableEnv = TableEnvironment.getTableEnvironment(env)
    tableEnv.registerDataSet("tmp_table", transDataset, 'name, 'num)
    val table = tableEnv.sqlQuery("select name,num from tmp_table")
 
    val rtnDataset = tableEnv.toDataSet[(String,Integer)](table)
    rtnDataset.print()
 
    env.execute("test hdfs textfile")
  }
}

3.Hive

2.在pom檔案中新增依賴：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
  <version>2.7.3</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-hadoop-fs</artifactId>
  <version>1.6.1</version>
</dependency>
<dependency>
  <groupId>com.jolbox</groupId>
  <artifactId>bonecp</artifactId>
  <version>0.8.0.RELEASE</version>
</dependency>
<dependency>
  <groupId>com.twitter</groupId>
  <artifactId>parquet-hive-bundle</artifactId>
  <version>1.6.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-exec</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-metastore</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-cli</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-common</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-service</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-shims</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hive.hcatalog</groupId>
  <artifactId>hive-hcatalog-core</artifactId>
  <version>1.2.2</version>
</dependency>
<dependency>
  <groupId>org.apache.thrift</groupId>
  <artifactId>libfb303</artifactId>
  <version>0.9.3</version>
  <type>pom</type>
</dependency>

3.在flink-lib中新增下面所有jar：

accumulo-core-1.6.0.jar derby-10.11.1.1.jar hive-serde-1.2.0.jar mail-1.4.1.jar
accumulo-fate-1.6.0.jar derbyclient-10.14.2.0.jar hive-service-1.2.0.jar maven-scm-api-1.4.jar
accumulo-start-1.6.0.jar eigenbase-properties-1.1.5.jar hive-shims-0.20S-1.2.0.jar maven-scm-provider-svn-commons-1.4.jar
accumulo-trace-1.6.0.jar flink-dist_2.11-1.6.1.jar hive-shims-0.23-1.2.0.jar maven-scm-provider-svnexe-1.4.jar
activation-1.1.jar flink-hadoop-compatibility_2.11-1.6.1.jar hive-shims-1.2.0.jar netty-3.7.0.Final.jar
ant-1.9.1.jar flink-python_2.11-1.6.1.jar hive-shims-common-1.2.0.jar opencsv-2.3.jar
ant-launcher-1.9.1.jar flink-shaded-hadoop2-uber-1.6.1.jar hive-shims-scheduler-1.2.0.jar oro-2.0.8.jar
antlr-2.7.7.jar geronimo-annotation_1.0_spec-1.1.1.jar hive-testutils-1.2.0.jar paranamer-2.3.jar
antlr-runtime-3.4.jar geronimo-jaspic_1.0_spec-1.0.jar httpclient-4.4.jar parquet-hadoop-bundle-1.6.0.jar
apache-curator-2.6.0.pom geronimo-jta_1.1_spec-1.1.1.jar httpcore-4.4.jar parquet-hive-bundle-1.6.0.jar
apache-log4j-extras-1.2.17.jar groovy-all-2.1.6.jar ivy-2.4.0.jar pentaho-aggdesigner-algorithm-5.1.5-jhyde.jar
asm-commons-3.1.jar guava-14.0.1.jar janino-2.7.6.jar php
asm-tree-3.1.jar guava-15.0.jar jcommander-1.32.jar plexus-utils-1.5.6.jar
avro-1.7.5.jar hamcrest-core-1.1.jar jdo-api-3.0.1.jar postgresql-42.0.0.jar
bonecp-0.8.0.RELEASE.jar hive-accumulo-handler-1.2.0.jar jetty-all-7.6.0.v20120127.jar py
calcite-avatica-1.2.0-incubating.jar hive-ant-1.2.0.jar jetty-all-server-7.6.0.v20120127.jar regexp-1.3.jar
calcite-core-1.2.0-incubating.jar hive-beeline-1.2.0.jar jline-2.12.jar servlet-api-2.5.jar
calcite-linq4j-1.2.0-incubating.jar hive-cli-1.2.0.jar joda-time-2.5.jar slf4j-log4j12-1.7.7.jar
curator-client-2.6.0.jar hive-common-1.2.0.jar jpam-1.1.jar snappy-java-1.0.5.jar
curator-framework-2.6.0.jar hive-contrib-1.2.0.jar json-20090211.jar ST4-4.0.4.jar
curator-recipes-2.6.0.jar hive-exec-1.2.0.jar jsr305-3.0.0.jar stax-api-1.0.1.jar
datanucleus-api-jdo-3.2.1.jar hive-hbase-handler-1.2.0.jar jta-1.1.jar stringtemplate-3.2.1.jar
datanucleus-api-jdo-3.2.6.jar hive-hcatalog-core-1.2.2.jar junit-4.11.jar super-csv-2.2.0.jar
datanucleus-core-3.2.10.jar hive-hwi-1.2.0.jar libfb303-0.9.2.jar tempus-fugit-1.1.jar
datanucleus-core-3.2.2.jar hive-jdbc-1.2.0.jar libthrift-0.9.2.jar velocity-1.5.jar
datanucleus-rdbms-3.2.1.jar hive-jdbc-1.2.0-standalone.jar log4j-1.2.16.jar xz-1.0.jar
datanucleus-rdbms-3.2.9.jar hive-metastore-1.2.0.jar log4j-1.2.17.jar zookeeper-3.4.6.jar

4.下載hive1.2.0版本（根據自己的需要來），將hive-site.xml拷貝一份到hive/conf目錄下；啟動hive thrift metastore

5.即可以讀取hive表：

package com.flinklearn.main
 
 
import com.flinklearn.models.Alert
import org.apache.flink.api.scala.ExecutionEnvironment
 
import org.apache.flink.api.scala._
import org.apache.hadoop.conf.Configuration
 
/**
 * Created by dell on 2018/10/25.
 */
object Main {
  def main(args:Array[String]):Unit = {
    val conf = new Configuration()
    conf.set("hive.metastore.local", "false")
    conf.set("hive.metastore.uris", "thrift://ip:9083")
 
    val env = ExecutionEnvironment.getExecutionEnvironment
    val dataset = env.createInput(new HCatInputFormat[Alert]("db", "tb", conf))
 
    dataset.first(10).print()
 
    env.execute("flink hive test")
  }
}

4.JDBC


package com.flinklearn.main;
 
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
 
 
/**
 * Created by dell on 2018/10/29.
 */
public class Main {
    public static void main(String[] args){
        try {
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            JDBCInputFormat inputFormat = JDBCInputFormat.buildJDBCInputFormat()
                    .setDrivername("org.postgresql.Driver")
                    .setDBUrl("jdbc:postgresql://ip:port/nsc")
                    .setUsername("username")
                    .setPassword("password")
                    .setQuery("select xx,yy from zz")
                    .setRowTypeInfo(new RowTypeInfo(TypeInformation.of(String.class), TypeInformation.of(String.class)))
                    .finish();
 
            DataSource source = env.createInput(inputFormat);
            source.print();
            env.execute("jdbc test");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

相關推薦

Apache Flink 各類關鍵資料格式讀取/SQL支援

目前事件歸併分為兩種，一種為實時的歸併，即基於Kafka內的資料進行歸併和事件生成；一種是週期性的歸併，即基於Hive中的資料進行資料的歸併和事件生成。 基於SQL歸併時Spark Streaming支援的輸入/輸出資料如下： 資料型別 Fl

Apache Flink-資料流之上的有狀態的計算

官網給出的Flink應用場景圖： 1.狀態計算。 2.從上圖看出的應用場景有？ -----事件驅動式的流處理 -----ETL管道 -----資料分析 3.一般性架構 從架構上來說與一般的流式架構沒有太大的不同，任務排程和資源管理可以放在我們熟悉的yarn上進

.nc格式資料的讀取

netcdf資料：.nc格式資料的讀取（使用excel4netcdf開啟NC資料） 讀取.nc檔案的方式好像比較多，本人也是剛開始接觸.nc檔案，著實焦頭爛額了一把。無意間看到了使用Excell外掛開啟.nc檔案，感覺挺好用，記錄在此，方便以後使用。 1、安裝Excell外掛 1

spark1.6使用：讀取本地外部資料，把RDD轉化成DataFrame，儲存為parquet格式,讀取csv格式

一、先開啟Hadoop和spark 略 二、啟動spark-shell spark-shell --master local[2] --jars /usr/local/src/spark-1.6.1-bin-hadoop2.6/libext/com.mysql.jdbc.Driver.j

關於TP3.2框架讀取Sql server中文欄位資料以及處理亂碼的一些小心得

最近要做一個專案，需要使用TP3.2框架，之前什麼也不會，就硬著頭皮上了，結果真的鬧了挺多emmmmmm挺低階的錯誤，就像SQL Server中文欄位的讀取，一開始我是照著讀取英文欄位的格式來寫的，在Model層裡 public function getAdminByUsername($userna

SQL server匯入資料格式為科學

問題描述：Excel匯入資料庫的時候，有一列資料是九位數字 如72020693 匯入以後預設為float型別 我要把它改成nvarchar型別，直接修改了以後，他就會變成7202+e07的形式 （因為已經改好了沒有具體的例子） 後來我就把這列(欄位名為CostC

OpenCV 以MJPEG的格式 讀取攝像頭資料

                    OpenCV   以MJPEG的格式  讀取攝像頭資料   可以使用opencv讀取。 原始碼： #include

python讀取sql裡面的指定資料列，並將其轉換成列表使用

程式碼如下： import pyodbc import pandas as pd import numpy as np conn = pyodbc.connect(r'DRIVER={SQL Server Native Client 10.0};SERVER=.;DATABASE=資料庫名字

pandas讀取完的csv資料格式

使用pandas讀取csv檔案讀取出來的資料其實還是有一定附加格式的，那就是行列標題。 In [18]: data1 = pd.read_csv('data.csv') In [19]: data1

cifar10資料格式以及讀取方式

cifar10 資料網站 http://www.cs.toronto.edu/~kriz/cifar.html 讀取下面的檔案 CIFAR-10 binary version (suitable for C programs) 162 MB c32a1d4ab5d03f1284b

Apache Flink SQL示例

本章講解如何使用Flink SQL進行批處理操作。 這裡我提取了2016年中超聯賽射手榜的資料，通過Flink SQL進行簡單的彙總。 1、源資料 這裡儲存為csv格式： 2、在pom中新增Table API的依賴： <depende

opencv 讀取 普通視訊、以YUV資料格式儲存的視訊 以及 圖片序列的方法

       在剛入門opencv的階段，讀取圖片以及視訊是作為初學者最需要掌握的方法。最近在上智慧視訊分析這門課程，通過幾次實驗課，發現針對不同的資料分別對應著不同的載入方式，特此開貼，記錄下來，一是可以與人分享，共同進步，二是權當學習筆記記錄下來，以供以後回顧。若有謬誤，

FFMPEG記憶體操作（二）從記憶體中讀取數及資料格式的轉換

   相關部落格列表：     在雷神的《最簡單的基於FFmpeg的記憶體讀寫例子（記憶體播放器）》中，它是設計回撥函式從輸入檔案中讀取資料。與FFMPEG 官方給出的avio_reading.c不同的是，雷神給的例子是當需要資料的時候，回撥函式才去從輸入檔案讀取資料，而av

KSQL，用於Apache Kafka的流資料SQL引擎

Apache Kafka是一個分散式的、分割槽的、多複本的日誌提交服務，使用Scala編寫，以可水平擴充套件和高吞吐率而被廣泛使用。Kafka最初是由LinkedIn開發，並於2011年初開源，目標是為實時資料處理提供一個統一、高通量、低等待的平臺。目前，越來越

matlab讀取寫入影象資料格式uint8,double

為了節省儲存空間，matlab為影象提供了特殊的資料型別uint8(8位無符號整數），以此方式儲存的影象稱作8點陣圖像。 imread把灰度影象存入一個8位矩陣，當為RGB影象時，就存入8位RGB矩陣中。 因此，matlab讀入影象的資料是uint8，而matlab中數

【雲星資料---Apache Flink實戰系列(精品版)】：Apache Flink高階特性與高階應用002-Flink的記憶體管理002

二、flink的記憶體管理機制 0.flink的記憶體劃分 1.flink在JVM的heap內有自己的記憶體管理空間。 2.在flink中記憶體被分為三個部分，分別是Unmanaged區域，

【雲星資料---Apache Flink實戰系列(精品版)】：Apache Flink高階特性與高階應用008-Slot和Parallelism的深入分析003

四、任務槽（task-slot）和槽共享（Slot Sharing） 1.任務槽（Task slot） 1.flink的TM就是執行在不同節點上的JVM程序（process）,這個程序會

【雲星資料---Apache Flink實戰系列(精品版)】：Apache Flink高階特性與高階應用001-Flink的記憶體管理001

一、基於JVM的大資料生態圈 1.bigdata on jvm 1.現在大多數開源大資料處理框架都是基於jvm的，像 Apache Hadoop,Apache Spark,Apache Hb

【雲星資料---Apache Flink實戰系列(精品版)】：Apache Flink實戰基礎0015--flink分散式部署0010

五、flink-on-yarn實現原理 1.通過配置資訊找到yarn a.Flink-Yarn-Client(FRC)讀取YARN_CONF_DIR,HADOOP_CONF_DIR或HAD

python讀取外部資料之讀取csv格式

20171204更新讀取csv檔案需要先open(如果路徑裡含有中文)：file = open(' csv 檔案完整路徑')df  = pd.read_csv(file)如果全英文路徑不需要open最近