Scala 实战

List(1, 9, 2, 4, 5) span (_ < 3)       // (List(1), List(9, 2, 4, 5))，碰到不符合就结束

List(1, 9, 2, 4, 5) partition (_ < 3)  // (List(1, 2), List(9, 4, 5))，扫描所有

List(1, 9, 2, 4, 5) splitAt 2          // (List(1, 9),List(2, 4, 5))，以下标为分割点

List(1, 9, 2, 4, 5) groupBy (5 < _)    // Map(false -> List(1, 2, 4, 5), true -> List(9))，分割成 Map 对象，以 Boolean 类型为 Key

Iterator

grouped

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.BigquerySparkSession._
val conf = new SparkConf()
val builder = SparkSession.builder().config(conf).enableHiveSupport()
val spark = builder.getOrCreateBigquerySparkSession()
val df = spark.sql("use db; select * from table")

val dataset = df.rdd.mapPartitions(iter => {

  val records = new util.LinkedList[String]()

  // 将每个 partition 中的多行数据，以 100 为长度作为一组，进行一次批处理
  iter.grouped(100)
    .foreach(rows => {
      rows.foreach(row => records.add(JSON.toJSONString(row, false)))
    })

  // 重新将一组处理后的数据，封装成 Iterator 类
  new AbstractIterator[String]() {

    var shouldOutput = true

    override def hasNext: Boolean = shouldOutput

    override def next(): String = {
      shouldOutput = false
      JSON.toJSONString(records, false)
    }
  }
})

// 反序列成 list 对象，再将 list 对象 flat 成单个元素，同时，过滤掉其中空白行
val filteredEmptyLine = dataset
  .filter(_ != null)
  .filter(!_.equals(""))
  .map(JSON.parseObject(_, classOf[util.LinkedList[String]]))
  .filter(_ != null)
  .filter(_.size() != 0)
  .flatMap(_.toArray)
  .filter(_ != null)
  .map(JSON.toJSONString(_, false))
  .filter(!_.equals(""))

Case class

和 class 的 8 个不同之处

// 定义
scala> case class Person(name:String, age:Int)
defined class Person

// 初始化
scala> val bj = Person("Benedict Jin", 18)
bj: Person = Person(Benedict Jin,18)

// 参数的访问权限都是 public
scala> bj.name
res0: String = Benedict Jin

scala> bj.age
res1: Integer = 18

// 更加简洁的 toString 打印
scala> bj.toString
res2: String = Person(Benedict Jin,18)

// 默认实现了 hashCode 和 equals 方法
scala> bj.hashCode
res3: Int = 1059149039

scala> Person("Benedict Jin", 18).hashCode
res4: Int = 1059149039

scala> Person("Benedict Jin", 18) == bj
res5: Boolean = true

// 默认实现了 java.io.Serializable 接口，支持序列化
scala> import java.io._
import java.io._

scala> val bos = new ByteArrayOutputStream
bos: java.io.ByteArrayOutputStream =

scala> val oos = new ObjectOutputStream(bos)
oos: java.io.ObjectOutputStream = java.io.ObjectOutputStream@62ddd21b

scala> oos.writeObject(bj)

scala> val obj = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray)
).readObject()
obj: Object = Person(Benedict Jin,18)

scala> Person("Benedict Jin", 18) == obj
res6: Boolean = true

// 支持 match
scala> bj match {
     |   case Person("Benedict Jin", 18) => println("matched")
     |   case _ => println("non-matched")
     | }
matched

// 默认继承 scala.Product 类，并实现了其中的方法
scala> bj.productArity
res7: Int = 2

scala> bj.productIterator.next()
res8: Any = Benedict Jin

scala> bj.productElement(1)
res9: Any = 18

scala> bj.productPrefix
res10: String = Person

打印 trait 中的字段

scala> trait T{
     |   var t: String = ""
     | }
defined trait T

// 正常情况下，不会打印 trait 中的字段
scala> case class Person(name:String)extends Serializablewith T
defined class Person

scala> Person("Benedict Jin")
res6: Person = Person(Benedict Jin)

// 通过 ScalaRunTime._toString(this) 来覆写 toString 方法，可以避免 case class 的 toString 方法被篡改
scala> import scala.runtime.ScalaRunTime
import scala.runtime.ScalaRunTime

scala> case class Person2(name:String)extends Serializablewith T{
     |   override def toString: String = ScalaRunTime._toString(this)
     | }
defined class Person2

scala> Person2("Benedict Jin").toString
res7: String = Person2(Benedict Jin)

// 重写 toString 来打印 trait 中的字段
scala> case class Person3(name:String)extends Serializablewith T{
     |   override def toString: String = s"Person3(${this.name},${this.t})"
     | }
defined class Person3

scala> val p3 = Person3("Benedict Jin")
p3: Person3 = Person3(Benedict Jin, )

scala> p3.t = "t"
p3.t: String = t

scala> p3.toString
res8: String = Person3(Benedict Jin, t)

单元测试

详见，《在不同场景下， 如何选择合适的 JVM 语言 》

常见问题

classOf 和 getClass 的区别

scala> class A
  defined class A

scala> val a = new A
  a: A = A@1d483de4

// A 的任意子类
scala> a.getClass
  res0: Class[_ <: A] = class A

// A 类本身
scala> classOf[A]
  res1: Class[A] = class A

// 两者是等价的
scala> a.getClass == classOf[A]
  res2: Boolean = true

// getClass 的返回值，是不能直接赋值给 Class[A] 的
scala> val c: Class[A] = a.getClass
  <console>:13: error: type mismatch;
   found   : Class[?0] where type ?0<: A
   required: Class[A]
  Note: ?0 <: A, but Java-defined class ClassisinvariantintypeT.
  You may wish to investigate a wildcard type suchas`_<: A`. (SLS 3.2.10)
         val c:Class[A] = a.getClass
                            ^

// 需要声明 Class[_ <: A] 才行，类似于  Java  里面的 Class<? extends A>
scala> val c: Class[_ <: A] = a.getClass
  c: Class[_ <: A] = class A

// 但是获取全限定名的返回值，却是一样的
scala> c.getName
  res3: String = A

scala> classOf[A].getName
  res4: String = A

稳定的标识符模式

// 如果不用 ` 反引号将 a / b / c 变量括起来，那么这些变量，其实就变成了指向 i 的别名，已经和 match 外层的 a / b / c 变量无关了
scala> def m(i: Int) = {
     |   val a = 3
     |   val b = 2
     |   val c = 1
     |   i match {
     |     case `a` => 0
     |     case `b` => -1
     |     case `c` => 4
     |     case _ => 2
     |   }
     | }
m: (i: Int)Int

scala> m(1)
res17: Int = 4

scala> m(2)
res18: Int = -1

scala> m(3)
res19: Int = 0

// 另外，使用反引号将变量括起来之后，scala 会在字节码层面做优化
/*
 0: iconst_3
 1: istore_2
 2: iconst_2
 3: istore_3
 4: iconst_1
 5: istore          4
 7: iload_1
 8: istore          5
10: iload_2
11: iload           5
13: if_icmpne       22
16: iconst_0
17: istore          6
19: goto            50
22: iload_3
23: iload           5
25: if_icmpne       34
28: iconst_m1
29: istore          6
31: goto            50
34: iload           4
36: iload           5
38: if_icmpne       47
41: iconst_4
42: istore          6
44: goto            50
47: iconst_2
48: istore          6
50: iload           6
52: ireturn
 */

Tips: Full code is here and here .

参考

• Why can’t a variable be a stable identifier?

在函数调用里面，对变量进行赋值

scala> val map = new java.util.LinkedHashMap[String, String]()
map: java.util.LinkedHashMap[String,String] = {}

scala> map.put("1", "1")
res0: String = null

scala> map.put("2", "2")
res1: String = null

scala> map.put("3", "3")
res2: String = null

scala> map
res3: java.util.LinkedHashMap[String,String] = {1=1, 2=2, 3=3}

scala> var s = "1 "
s: String = "1 "

scala> map.containsValue(s)
res4: Boolean = false

// 虽然已经将 s.trim 赋值给 s 变量了，但是传入 map.containsValue 方法的仍然是未进行 trim 操作之前的 s 变量
// 这里和 Java 是不一样的，Java 会将已经赋值之后的变量值，传入到方法中
scala> map.containsValue(s = s.trim)
res5: Boolean = false

scala> s
res6: String = 1

scala> s.equals("1")
res7: Boolean = true

Java 的 Lambda 表达式转为 Scala 的 Function

Java 的 Lambda 表达式

final static ThreadLocal<String> BLOG_ID = ThreadLocal.withInitial(() -> "null");

反面示例

// 直接填入 lambda 表达式，IDE 是不会报错的，但是会在代码编译阶段报错
private val BLOG_ID: ThreadLocal[String] = ThreadLocal.withInitial(() -> "null")

// 即便把 function 单独拿出来作为 Supplier 变量进行申明，仍然会出现编译错误
val blogIdFunc: java.util.function.Supplier[String] = () => "null"

// 具体编译报错，如下：
val func: java.util.function.Supplier[String] = () => "null"
 error: type mismatch;
 found   : () => String
 required: java.util.function.Supplier[String]
       val func: java.util.function.Supplier[String] = () => "null"
                                                          ^

正面示例

// 写成 new Supplier[T]{...} 的写法之后，可以完成编译
// 但是，编译器仍然会提示，可以将代码优化为 () => "null"
val blogIdFunc: java.util.function.Supplier[String] = new Supplier[String] { override def get(): String = "null" }
private val BLOG_ID: ThreadLocal[String] = ThreadLocal.withInitial(blogIdFunc)

补充

// 如果该部分代码仍然需要运行在 JDK7 及以下版本的 JVM 环境中，则可以改成普通的 "初始化 ThreadLocal" 方式
private val BLOG_ID: ThreadLocal[String] = new ThreadLocal[String]() {
  override def initialValue(): String = "null"
}

如何传递变长参数

描述

scala> def detail(d: Any*): Unit = println("%s_%s".format(d))
detail: (d: Any*)Unit

scala> detail("a", "b")
java.util.MissingFormatArgumentException: Format specifier '%s'
  at java.util.Formatter.format(Formatter.java:2519)
  at java.util.Formatter.format(Formatter.java:2455)
  at java.lang.String.format(String.java:2940)
  at scala.collection.immutable.StringLike$class.format(StringLike.scala:318)
  at scala.collection.immutable.StringOps.format(StringOps.scala:29)
  at .detail(<console>:11)
  ... 32 elided

解决

// 在函数中，想要将传入的变长参数，保持成多个参数的特性，传递下去的话，需要声明 : _*
scala> def detail(d: Any*): Unit = println("%s_%s".format(d: _*))
detail: (d: Any*)Unit

scala> detail("a", "b")
a_b