添雨

1
2
3
4
5
6

vi /etc/profile
# 插入如下语句
export SPARK_HOME=/usr/local/spark-2.1.0
export PATH=$PATH:$SPARK_HOME/bin
# 执行语句，使环境变量生效
source /etc/profile

1
2

val conf = new SparkConf().setAppName(appName).setMaster(master)
new SparkContext(conf)

1

$ ./bin/spark-shell --master local[4]

1

$ ./bin/spark-shell --master local[4] --jars code.jar

1

$ ./bin/spark-shell --master local[4] --packages "org.example:example:0.0.1"

1
2
3

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

1

lineLengths.persist()

1

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

1
2
3
4
5

val accum = sc.longAccumulator("My Accumulator")

sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))

accum.value //Long  = 10

1
2
3
4
5
6
7
8

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("Spark SQL示例")
.config("spark.some.config.option", "some-value")
.getOrCreate()

//  引用可将RDD转换为DataFrames的隐式转换
import spark.implicits._

1
2
3

val df = spark.read.json("examples/src/main/resources/people.json")
//  显示DataFrame的内容
df.show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

// This import is needed to use the $-notation
import spark.implicits._
// Print the schema in a tree format
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// Select only the "name" column
df.select("name").show()
// +-------+
// |   name|
// +-------+
// |Michael|
// |   Andy|
// | Justin|
// +-------+

// Select everybody, but increment the age by 1
df.select($"name", $"age" + 1).show()
// +-------+---------+
// |   name|(age + 1)|
// +-------+---------+
// |Michael|     null|
// |   Andy|       31|
// | Justin|       20|
// +-------+---------+

// Select people older than 21
df.filter($"age" > 21).show()
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+

// Count people by age
df.groupBy("age").count().show()
// +----+-----+
// | age|count|
// +----+-----+
// |  19|    1|
// |null|    1|
// |  30|    1|
// +----+-----+

1
2
3

df.createOrReplaceTempView("people");
val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show();

1
2
3
4
5
6
7

//  将 DF 注册到系统保留的数据数 global_temp
df.createGlobalTempView("people")
// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()

//  全局临时视图是跨会话的，所以新创建一个会话
spark.newSession().sql("SELECT * FROM global_temp.people").show()

1
2
3
4
5

case class Person(name: String, age: Long)

//  为案例类创建编码器
val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS.show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

//  读取并转换数据
val peopleDF = spark.sparkContext
  .textFile("src/main/resources/people.csv")
  .map(_.split(","))
  .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
  .toDF()
//  创建或替换临时视图
peopleDF.createOrReplaceTempView("people")
val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")

//  使用索引值访问
teenagersDF.map(teenager => "Name:" + teenager(0)).show()

//  使用字段名称访问
teenagersDF.map(teenager => "Name:" + teenager.getAs[String]("name")).show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import spark.implicits._
import org.apache.spark.sql.types._

//  创建一个RDD
val peopleRDD = spark.sparkContext.textFile("src/main/resources/people.csv")
//  使用字符串作为模式创建
val schemaString = "name age"
//  使用字符串的模式创建模式
val fields = schemaString.split(" ")
  .map(fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)

//  将RDD转换成 Rows
val rowRDD = peopleRDD
  .map(_.split(","))
  .map(attributes => Row(attributes(0), attributes(1).trim))

//  使用RDD应用模式
val peopleDF = spark.createDataFrame(rowRDD, schema)

//  创建临时视图
peopleDF.createOrReplaceTempView("people")

//  此时SQL可以在临时视图上运行
val results = spark.sql("SELECT name FROM people")

//  输出结果
results.map(attributes => "Name: " + attributes(0)).show()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

object MyAverage extends UserDefinedAggregateFunction {
override def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)

override def bufferSchema: StructType = {
  StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)
}

override def dataType: DataType = DoubleType

override def deterministic: Boolean = true

override def initialize(buffer: MutableAggregationBuffer): Unit = {
  buffer(0) = 0L
  buffer(1) = 0L
}

override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
  if (!input.isNullAt(0)) {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }
}

override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
  buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
  buffer1(1) = buffer1.getLong(0) + buffer2.getLong(0)
}

override def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)
}

def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("User Defined Test").master("local")
  .getOrCreate()

//  注册自定义聚合类
spark.udf.register("myAverage", MyAverage)

val df = spark.read.json("src/main/resources/employees.json")
df.createOrReplaceTempView("employees")
df.show()

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")
result.show()
}

1
2
3
4
5

val usersDF = spark.read.format("json").load("src/main/resources/users.parquet")
usersDF.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

//载入时使用
spark.read.load("namesAndFavColors.parquet")

1
2
3
4
5

val peopleDFCsv = spark.read.format("csv")
    .option("sep", ";")
    .option("inferSchema", "true")
    .option("title", true)
    .load("src/main/resources/people.csv")

1

val sqlDF = spark.sql("SELECT * FROM parquest.`src/main/resources/users.parquet`")

1
2

df.select("name", 
"age").write.format("parquet").mode(SaveMode.ErrorIfExists).save("peopleSave.parquet")

1
2
3

dependencies {
    compile ("mysql:mysql-connector-java:5.1.46")
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

val spark = SparkSession.builder()
    .appName("Spark JDBC SQL示例")
    .master("local")
    .config("spark.some.config.option", "some-value")
    .getOrCreate()
val prop = new Properties
prop.setProperty("user", "root")
prop.setProperty("password", "root")

val df = spark.read.jdbc("jdbc:mysql://localhost:3306/yunpos", "cm_order", prop)
//或者直接使用：val df = spark.read.jdbc("jdbc:mysql://localhost:3306/yunpos?user=root&password=root", "cm_order", new Properties)

println(df.count())

1

spark.conf.getAll.foreach(println)

1

spark.conf.set("hive.metastore.uris", "thrift://bigdata3:9083")

1
2
3
4

//  导入 SQL 相关的方法
import org.apache.spark.sql.functions._
//  导入 可将RDD隐式转换为DF的操作集合
import spark.implicits._

1

val DF = spark.table("table_name")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

val spark = SparkSession.builder()
  .appName("Spark MySql Score 示例")
  .master("local")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()
val df = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/yunpos?user=root&password=root", "sc_order", new Properties)
//  创建临时视图
df.createTempView("order")
//  直接使用SQL语句进行统计
val upDF = spark.sql("SELECT upstream_name, count(id) as count FROM order GROUP BY upstream_name")
upDF.show()
//  保存数据
upDF.write.format("json").saveAsTable("TW_SCORE_ORDER_W_X")
spark.stop()

1

joinDF1.join(joinDF2)

1

joinDF1.join(joinDF2, "id")

1

joinDF1.join(joinDF2, Seq("id", "name"))

1

joinDF1.join(joinDF2, Seq("id", "name"), "inner")

1

joinDF1.join(joinDF2, joinDF1("id") === joinDF2("t1_id"))

1

joinDF1.join(joinDF2, joinDF1("id") === joinDF2("t1_id"), "inner")

1
2
3
4

 val logDF = spark.read
      .option("header", "true")
      .option("encoding", "gbk")
      .csv("src/main/resources/log.csv")

1
2
3
4

val iskwDF = logDF
      .withColumn("stepTemp", logDF.col("step").cast(IntegerType))
      .drop("step")
      .withColumnRenamed("stepTemp", "step")

1
2
3
4

//  等于一个值
df.filter($"iskw" === "1")
//  小于一个值
df.filter($"iskw" < 2) 

1
2
3
4
5
6

import sqlContext.implicits._
import org.apache.spark.sql.functions._
// 方法一
df.agg(sum("isa")).first.get(0)
// 方法二
df.select(col("steps")).rdd.map(_(0).asInstanceOf[Int]).reduce(_+_)

1

df.repartition(1).write.option("header", "true").csv("src/main/resources/result.csv")

1

val list = df.rdd.map(r => r(0).asInstanceOf[String]).collect.toList

1

val resultUsrList = filter1DF.filter($"usrid".isin(filterIsaAsOneList: _*))

1
2

import sqlContext.implicits._ 
val df = Seq( (1, "First Value", java.sql.Date.valueOf("2010-01-01")), (2, "Second Value", java.sql.Date.valueOf("2010-02-01")) ).toDF("int_column", "string_column", "date_column")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

/ sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// this is used to implicitly convert an RDD to a DataFrame.
import sqlContext.implicits._

// Define the schema using a case class.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,
// you can use custom classes that implement the Product interface.
case class Person(name: String, age: Int)

// Create an RDD of Person objects and register it as a table.
val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.registerTempTable("people")

// 使用 sqlContext 执行 sql 语句.
val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// 注：sql()函数的执行结果也是DataFrame，支持各种常用的RDD操作.
// The columns of a row in the result can be accessed by ordinal.
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

val spark = SparkSession.builder().appName("YuewenAdReport")
.enableHiveSupport()
.config("hive.exec.dynamic.partition", "true")
.config("hive.exec.dynamic.partition.mode", "nonstrict") // 增加对分区操作权限
.getOrCreate()

//以下代码证明无效，会全盘覆盖
//val DF = spark.read.table("test");
//DF.write.mode("overwrite").format("hive").insertInto("tableName")

// 使用sql时：

spark.sql("select *, date_format(to_date(create_time, 'yyyy-MM-dd'), 'yyyyMMdd') as pt_day from table").createOrReplaceTempView("tempTable")
spark.sql("INSERT OVERWRITE TABLE demo_tab PARTITION (pt_day) SELECT stadt, geograph_breite, id, t.country FROM demo_stg t;)

1
2
3

spark-shell --conf "hive.exec.dynamic.partition=true" --conf "hive.exec.dynamic.partition.mode=nonstrict"

df.write.mode("overwrite").format("hive").insertInto("tableName")

1

spark.sql("ALTER TABLE ds1505_tbl_sms_log DROP IF EXISTS PARTITION(pt_day = '20190101')").show()

1
2
3
4
5

import spark.implicits._
val a = Seq("1", "2", "3", "4", "5", "6", "7", "8", "9").toDF("test")
a.show(false)
a.printSchema
a.orderBy(rand()).show

1
2
3
4

import spark.implicits._

df.select($"*" +: Seq("A", "B", "C").map(c => 
  sum(c).over(Window.partitionBy("ID").orderBy("time")).alias(s"cum$c")): _*)

1
2
3
4
5
6
7
8
9

import org.apache.spark.sql.{Column, DataFrame}

/**
 * @param cols a sequence of columns to transform
 * @param df an input DataFrame
 * @param f a function to be applied on each col in cols
 */
def withColumns(cols: Seq[String], df: DataFrame, f: String => Column) =
  df.select($"*" +: cols.map(c => f(c)): _*)

1
2
3

Seq("A", "B", "C").foldLeft(df)((df, c) =>
  df.withColumn(s"cum$c",  sum(c).over(Window.partitionBy("ID").orderBy("time"))))

1
2
3
4
5
6
7
8

/**
 * @param cols a sequence of columns to transform
 * @param df an input DataFrame
 * @param f a function to be applied on each col in cols
 * @param name a function mapping from input to output name.
 */def withColumns(cols: Seq[String], df: DataFrame, 
    f: String =>  Column, name: String => String = identity) =
  cols.foldLeft(df)((df, c) => df.withColumn(name(c), f(c)))

1
2
3
4
5
6
7
8

val sqlContext = new SQLContext(sc)
import sqlContext.implicits._ // for `toDF` and $""
import org.apache.spark.sql.functions._ // for `when`

val df = sc.parallelize(Seq((4, "blah", 2), (2, "", 3), (56, "foo", 3), (100, null, 5)))
    .toDF("A", "B", "C")

val newDf = df.withColumn("D", when($"B".isNull or $"B" === "", 0).otherwise(1))

1

hadoop fs -put file.csv /user/tangf/

1
2
3
4

val table = spark.read.option("header", 
"true").csv("hdfs://bigdata4:8020/user/tangf/file.csv")

table.write.mode("overwrite").saveAsTable("table_name")

1

val table = spark.read.option("header", "false").option("delimiter", "`").option("encoding", "gbk").schema("ID BIGINT, MID STRING, WeekIndex INT, Mobile STRING, TaskId INT, InterfaceID INT, Content STRING, UserID INT, UpdateTime TIMESTAMP, AddTime TIMESTAMP, Status INT, SubmitErrorCode STRING, ReportErrorCode STRING, MobileArea STRING, ActualCount INT, ExtNo STRING, AccessCode STRING, DeductFlag INT").csv("hdfs://bigdata4:8020/user/test/file.csv")

1
2
3
4
5
6
7

table.createOrReplaceTempView("table")

val saveTable = spark.sql("select *, date_format(to_date(AddTime, 'yyyy-MM-dd'), 'yyyyMMdd') as pt_day from table")

saveTable.write.mode("overwrite").partitionBy("pt_day").saveAsTable("imissms_large_black")
//分区重载写入，以防覆盖整个表（2019-10-24 未测试）
spark.sql("insert overwrite tableA partition(pt_day) select * from table")

1
2

val table = spark.read.table("table_name")
table.write.mode("overwrite").option("timestampFormat", "yyyy/MM/dd HH:mm:ss ZZ").option("header", "false").csv("/user/test/table_name")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object CreditConvert {
  def main(args: Array[String]): Unit = {
     val spark = SparkSession.builder()
      .appName("Spark JDBC CSV")
      .master("local")
      .getOrCreate()
    import spark.implicits._
    //  注册获取当前处理的文件名称并获取从中获取日期字符串

    val get_file_name = udf((path: String) => "\\d{8}".r.findFirstIn(path).get.replaceAll("(\\d{4})(\\d{2})(\\d{2})", "$1-$2-$3"))
    val fileName = "F:\\test\\*.txt"

    val phoneDF = spark.read.text(fileName)
      .map(line => line.mkString.split(" {4}"))
      .flatMap(arr => {
        val phone = arr(0)
        val is_in = arr(1)
        val words = phone.split(" ")
        words.map(word => (word, is_in))
      }).withColumn("date", get_file_name(input_file_name())).toDF("phone", "is_in", "date")
    phoneDF.printSchema()
    phoneDF.show(false)
    phoneDF.repartition(1).write
      .option("header","true")
      .option("timestampFormat", "yyyy/MM/dd HH:mm:ss ZZ") //解决 java.lang.IllegalArgumentException: Illegal pattern component: XXX 报错问题
      .csv("src/main/resources/test_20190906.csv")
  }
}

1
2
3
4

select 
  from_unixtime(1508673584)                    as fut,
  cast(from_unixtime(1508673584) as date)      as futAsDate,
  cast(from_unixtime(1508673584) as timestamp) as futAsTimestamp;

1
2

-- 获取分组下的最新的数据。根据CreateTime desc 获取最新
select * from (select uuid, state, firstCard, CreateTime, cardProduct, ROW_NUMBER() over(partition by uuid order by CreateTime desc) as rn from ds13457_PuFaCredit ) a where a.rn = 1 limit 100

1

spark.sql("desc formatted beacon_web_toutiao_report").toDF.filter('col_name === "Location").collect()(0)(1).toString

1

spark.read.table("ds3147_tblPhoneTask").inputFiles

1
2
3
4
5
6
7
8

version=1.0.0-SNAPSHOT
scalaVersion=2.11.8
slf4jVersion=1.7.21
logbackVersion=1.1.7
sparkVersion=2.3.0

title=SparkLearn
group=com.tangfan.spark.example

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

plugins {
    id 'java'
    id 'scala'
    id 'idea'
    id 'application'
}
dependencies {
    compile "org.scala-lang:scala-library:${scalaVersion}"
    compile "org.scala-lang:scala-reflect:${scalaVersion}"
    compile "org.scala-lang:scala-compiler:${scalaVersion}"

    compile "org.apache.spark:spark-mllib_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-sql_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-streaming-flume-assembly_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-graphx_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-launcher_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-catalyst_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-streaming_2.11:${sparkVersion}"
    compile "org.apache.spark:spark-core_2.11:${sparkVersion}"
    compile group: 'commons-io', name: 'commons-io', version: '2.5'
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

plugins {
    id 'com.github.johnrengelman.shadow' version '2.0.4'
}

shadowJar {
    classifier = 'shadow'
    dependencies {
        //  只将某些包打入进去
        include(dependency("com.xx:xx:1.0.0"))
    }
}

1

$ ./bin/spark-submit --class path.to.your.CLass --master yarn --deploy-mode cluster [options] <app jar> [app options]

1
2
3
4
5
6
7
8

$ ./bin/spark-submit --class org.apache.spark.examples.SparkPi \
    --master yarn \
    --deploy-mode cluster \
    --driver-memory 4g \
    --executor-memory 2g \
    --queue thequeue \
    xx.jar \
    10

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

spark-shell --executor-memory 15G --executor-cores 8

spark-shell \
--executor-memory 4G \
--num-executors 8 \
--executor-cores 4 \
--driver-memory 10G \
--conf "spark.defalut.parallelism=1000" \
--conf "spark.storage.memoryFraction=0.5" \
--conf "spark.shuffle.memoryFraction=0.3"

1
2
3

spark-shell \
--conf "hive.exec.dynamic.partition=true"
--conf "hive.exec.dynamic.partition.mode=nonstrict"

1

spark-submit --master yarn --deploy-mode client --driver-memory 1G --executor-memory 2G --executor-cores 4 --num-executors 3 --queue dev

1

spark-submit --class 应用主类 --jars 执行过程中使用到的JAR包

1

spark.time(spark.sql("xxx").show)

1
2
3
4
5
6
7
8

scala> spark.time(spark.sql("select count(1) as c from ds1505_tbl_sms_log").show)
+---------+
|        c       |
+---------+
|129877561|
+---------+

Time taken: 7834 ms

1
2
3
4
5

spark-submit --class com.path.to.class.InitialContactDriver \
--driver-java-options "-Dlog4j.configuration=file:log4j.properties" \
--conf "spark.executor.extraJavaOptions=Dlog4j.configuration=file:log4j.properties" \
--driver-class-path /home/cloudera/SNAPSHOT.jar \
--master yarn    /home/cloudera/SNAPSHOT.jar

1

spark.time(spark.sql("select count(1) as c from ds1505_tbl_sms_log").show)

1

--executor-memory 15G --executor-cores 8

1
2
3
4
5
6

//读取1：直接使用Files读取
new String(Files.readAllBytes(Paths.get(url)), "utf-8")

//读取2：使用Spark读取
val wholeFiles = spark.sparkContext.wholeTextFiles(tagsJsonUrl)
wholeFiles.collect().head._2

1

val url = System.getenv("SPARK_YARN_STAGING_DIR") + s"/$url"