import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, _}
import scala.collection.immutable.{Queue, Stack}
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}object DataSource001 {def main(args: Array[String]): Unit = {val env = ExecutionEnvironment.getExecutionEnvironment//0.用element创建DataSet(fromElements)val ds0: DataSet[String] = env.fromElements("spark", "flink")ds0.print()//1.用Tuple创建DataSet(fromElements)val ds1: DataSet[(Int, String)] = env.fromElements((1, "spark"), (2, "flink"))ds1.print()//2.用Array创建DataSetval ds2: DataSet[String] = env.fromCollection(Array("spark", "flink"))ds2.print()//3.用ArrayBuffer创建DataSetval ds3: DataSet[String] = env.fromCollection(ArrayBuffer("spark", "flink"))ds3.print()//4.用List创建DataSetval ds4: DataSet[String] = env.fromCollection(List("spark", "flink"))ds4.print()//5.用List创建DataSetval ds5: DataSet[String] = env.fromCollection(ListBuffer("spark", "flink"))ds5.print()//6.用Vector创建DataSetval ds6: DataSet[String] = env.fromCollection(Vector("spark", "flink"))ds6.print()//7.用Queue创建DataSetval ds7: DataSet[String] = env.fromCollection(Queue("spark", "flink"))ds7.print()//8.用Stack创建DataSetval ds8: DataSet[String] = env.fromCollection(Stack("spark", "flink"))ds8.print()//9.用Stream创建DataSet（Stream相当于lazy List，避免在中间过程中生成不必要的集合）val ds9: DataSet[String] = env.fromCollection(Stream("spark", "flink"))ds9.print()//10.用Seq创建DataSetval ds10: DataSet[String] = env.fromCollection(Seq("spark", "flink"))ds10.print()//11.用Set创建DataSetval ds11: DataSet[String] = env.fromCollection(Set("spark", "flink"))ds11.print()//12.用Iterable创建DataSetval ds12: DataSet[String] = env.fromCollection(Iterable("spark", "flink"))ds12.print()//13.用ArraySeq创建DataSetval ds13: DataSet[String] = env.fromCollection(mutable.ArraySeq("spark", "flink"))ds13.print()//14.用ArrayStack创建DataSetval ds14: DataSet[String] = env.fromCollection(mutable.ArrayStack("spark", "flink"))ds14.print()//15.用Map创建DataSetval ds15: DataSet[(Int, String)] = env.fromCollection(Map(1 -> "spark", 2 -> "flink"))ds15.print()//16.用Range创建DataSetval ds16: DataSet[Int] = env.fromCollection(Range(1, 9))ds16.print()//17.用fromElements创建DataSetval ds17: DataSet[Long] =  env.generateSequence(1,9)ds17.print()}
}

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment,_}object DataSource002 {def main(args: Array[String]): Unit = {val env = ExecutionEnvironment.getExecutionEnvironment//1.读取本地文本文件,本地文件以file://开头val ds1: DataSet[String] = env.readTextFile("file:///Applications/flink-1.1.3/README.txt")ds1.print()//2.读取hdfs文本文件，hdfs文件以hdfs://开头,不指定master的短URLval ds2: DataSet[String] = env.readTextFile("hdfs:///input/flink/README.txt")ds2.print()//3.读取hdfs CSV文件,转化为tupleval path = "hdfs://qingcheng11:9000/input/flink/sales.csv"val ds3 = env.readCsvFile[(String, Int, Int, Double)](filePath = path,lineDelimiter = "\n",fieldDelimiter = ",",lenient = false,ignoreFirstLine = true,includedFields = Array(0, 1, 2, 3))ds3.print()//4.读取hdfs CSV文件,转化为case classcase class Sales(transactionId: String, customerId: Int, itemId: Int, amountPaid: Double)val ds4 = env.readCsvFile[Sales](filePath = path,lineDelimiter = "\n",fieldDelimiter = ",",lenient = false,ignoreFirstLine = true,includedFields = Array(0, 1, 2, 3),pojoFields = Array("transactionId", "customerId", "itemId", "amountPaid"))ds4.print()}
}

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration/*** 递归读取hdfs目录中的所有文件，会遍历各级子目录*/
object DataSource003 {def main(args: Array[String]): Unit = {val env = ExecutionEnvironment.getExecutionEnvironment// create a configuration objectval parameters = new Configuration// set the recursive enumeration parameterparameters.setBoolean("recursive.file.enumeration", true)// pass the configuration to the data sourceval ds1 = env.readTextFile("hdfs:///input/flink").withParameters(parameters)ds1.print()}
}