for (arg <- args)

for (i <- 0 until 10)

val lines1 = sc.textFile("file:///usr/local/spark/mycode/rdd/word.txt")

val lines2 = sc.textFile("hdfs://localhost:9000/user/hadoop/word.txt")

val array = Array(1,2,3,4,5)
val rdd = sc.parallelize(array)

val list = List("Hadoop","Spark","Hive")
val rdd = sc.parallelize(list)
rdd.cache()   //会调用persist(MEMORY_ONLY)，但是，语句执行到这里，并不会缓存rdd，这是rdd还没有被计算生成
println(rdd.count())   //第一次行动操作，触发一次真正从头到尾的计算，这时才会执行上面的rdd.cache()，把这个rdd放到缓存中
println(rdd.collect().mkString(","))   //第二次行动操作，不需要触发从头到尾的计算，只需要重复使用上面缓存中的rdd

val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar.value  //Array(1, 2, 3)

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
scala> accum.value
res1: Long = 10

object MyFunctions {
    def func1(s: String): String = { ... }
}
myRdd.map(MyFunctions.func1)

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.linalg.{Vector, Vectors}
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.sql.Row
 
// Prepare training data from a list of (label, features) tuples.
val training = spark.createDataFrame(Seq(
    (1.0, Vectors.dense(0.0, 1.1, 0.1)),
    (0.0, Vectors.dense(2.0, 1.0, -1.0)),
    (0.0, Vectors.dense(2.0, 1.3, 1.0)),
    (1.0, Vectors.dense(0.0, 1.2, -0.5))
)).toDF("label", "features")
 
// Create a LogisticRegression instance. This instance is an Estimator.
val lr = new LogisticRegression()
// Print out the parameters, documentation, and any default values.
println("LogisticRegression parameters:\n" + lr.explainParams() + "\n")
 
// We may set parameters using setter methods.
lr.setMaxIter(10)
    .setRegParam(0.01)
 
// Learn a LogisticRegression model. This uses the parameters stored in lr.
val model1 = lr.fit(training)
// Since model1 is a Model (i.e., a Transformer produced by an Estimator),
// we can view the parameters it used during fit().
// This prints the parameter (name: value) pairs, where names are unique IDs for this
// LogisticRegression instance.
println("Model 1 was fit using parameters: " + model1.parent.extractParamMap)
 
// We may alternatively specify parameters using a ParamMap,
// which supports several methods for specifying parameters.
val paramMap = ParamMap(lr.maxIter -> 20)
    .put(lr.maxIter, 30)    // Specify 1 Param. This overwrites the original maxIter.
    .put(lr.regParam -> 0.1, lr.threshold -> 0.55)    // Specify multiple Params.
 
// One can also combine ParamMaps.
val paramMap2 = ParamMap(lr.probabilityCol -> "myProbability")    // Change output column name.
val paramMapCombined = paramMap ++ paramMap2
 
// Now learn a new model using the paramMapCombined parameters.
// paramMapCombined overrides all parameters set earlier via lr.set* methods.
val model2 = lr.fit(training, paramMapCombined)
println("Model 2 was fit using parameters: " + model2.parent.extractParamMap)
 
// Prepare test data.
val test = spark.createDataFrame(Seq(
    (1.0, Vectors.dense(-1.0, 1.5, 1.3)),
    (0.0, Vectors.dense(3.0, 2.0, -0.1)),
    (1.0, Vectors.dense(0.0, 2.2, -1.5))
)).toDF("label", "features")
 
// Make predictions on test data using the Transformer.transform() method.
// LogisticRegression.transform will only use the 'features' column.
// Note that model2.transform() outputs a 'myProbability' column instead of the usual
// 'probability' column since we renamed the lr.probabilityCol parameter previously.
model2.transform(test)
    .select("features", "label", "myProbability", "prediction")
    .collect()
    .foreach { case Row(features: Vector, label: Double, prob: Vector, prediction: Double) =>
        println(s"($features, $label) -> prob=$prob, prediction=$prediction")
    }

val spark = SparkSession.builder().appName("any").getOrCreate()
import spark.implicits._  //必须导入隐式转换

case class Person(name: String, age: Long)  //定义一个case class

val peopleDF = spark.sparkContext.textFile("file:///usr/local/spark/examples/src/main/resources/people.txt")
                                .map(_.split(","))
                                .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
                                .toDF()

//生成 RDD
val lines = spark.sparkContext.textFile("hdfs://10.30.30.146:9000/input/emp.csv").map(_.split(","))
//定义schema：StructType
val myschema = StructType(List(StructField("empno", DataTypes.IntegerType)
                                ,StructField("ename", DataTypes.StringType, nullable = true)
                                ,StructField("job", DataTypes.StringType)
                                ,StructField("mgr", DataTypes.StringType)
                                ,StructField("hiredate", DataTypes.StringType)
                                ,StructField("sal", DataTypes.IntegerType)
                                ,StructField("comm", DataTypes.StringType)
                                ,StructField("deptno", DataTypes.IntegerType)))
//把读入的每一行数据映射成一个个Row
val rowRDD = lines.map(x=>Row(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt))
//使用SparkSession.createDataFrame创建表
val df = spark.createDataFrame(rowRDD,myschema)

val peopleDF = spark.read.format("json").load("file:///usr/local/spark/examples/src/main/resources/people.json")
peopleDF.select("name", "age").write.format("csv").save("file:///usr/local/spark/mycode/newpeople.csv")

scala>empDF.createOrReplaceTempView("emp")
scala>spark.sql("select * from emp").show
scala>spark.sql("select * from emp where deptno=10").show

val df = spark.read.json("examples/src/main/resources/people.json")
df.select("name")
df.select($"name", $"age" + 1)
df.filter($"age" > 21)
df.groupBy("age").count()

val ageCol = people("age")
val futureAgeCol = people("age") + 10

df("columnName")            // On a specific `df` DataFrame.
col("columnName")           // A generic column no yet associated with a DataFrame.
col("columnName.field")     // Extracting a struct field
col("`a.column.with.dots`") // Escape `.` in column names.
$"columnName"               // Scala short hand for a named column.

people.filter(people.col("age").gt(30))
    .join(department, people.col("deptId").equalTo(department.col("id")))
    .groupBy(department.col("name"), people.col("gender"))
    .agg(avg(people.col("salary")), max(people.col("age")));