sparksql读取数据库问题

发表于 2019-04-18 | 分类于 sparksql |

| 字数统计: 110 | 阅读时长 ≈ 1

spark jdbc方式读取数据库，每个excutor都会去拉数据？

说明：该问题来源spark技术分享朋友群

spark.read("jdbc")
  .option("url", url)
  .option("dbtable", "pets")
  .option("user", user)
  .option("password", password)
  .option("numPartitions", 10)
  .option("partitionColumn", "owner_id")
  .option("lowerBound", 1)
  .option("upperBound", 10000)
  .load()

这种一般会转换为

SELECT * FROM pets WHERE owner_id >= 1 and owner_id < 1000
SELECT * FROM pets WHERE owner_id >= 1000 and owner_id < 2000
SELECT * FROM pets WHERE owner_id >= 2000 and owner_id < 3000
...

antlr4学习笔记

发表于 2019-04-18 | 分类于 antlr |

| 字数统计: 1,782 | 阅读时长 ≈ 8

antlr介绍

注意：文字来自《antlr4权威指南》
ANTLR 语言识别的一个工具 (ANother Tool for Language Recognition ) 是一种语言工具，它提供了一个框架，可以通过包含 Java, C++, 或 C# 动作（action）的语法描述来构造语言识别器，编译器和解释器。计算机语言的解析已经变成了一种非常普遍的工作，在这方面的理论和工具经过近 40 年的发展已经相当成熟，使用 Antlr 等识别工具来识别，解析，构造编译器比手工编程更加容易，同时开发的程序也更易于维护。
语言识别的工具有很多种，比如大名鼎鼎的 Lex 和 YACC，Linux 中有他们的开源版本，分别是 Flex 和 Bison。在 Java 社区里，除了 Antlr 外，语言识别工具还有 JavaCC 和 SableCC 等。
和大多数语言识别工具一样，Antlr 使用上下文无关文法描述语言。最新的 Antlr 是一个基于 LL(*) 的语言识别器。在 Antlr 中通过解析用户自定义的上下文无关文法，自动生成词法分析器 (Lexer)、语法分析器 (Parser) 和树分析器 (Tree Parser)。

antlr 的应用

编程语言处理

识别和处理编程语言是 Antlr 的首要任务，编程语言的处理是一项繁重复杂的任务，为了简化处理，一般的编译技术都将语言处理工作分为前端和后端两个部分。其中前端包括词法分析、语法分析、语义分析、中间代码生成等若干步骤，后端包括目标代码生成和代码优化等步骤。

Antlr 致力于解决编译前端的所有工作。使用 Anltr 的语法可以定义目标语言的词法记号和语法规则，Antlr 自动生成目标语言的词法分析器和语法分析器；此外，如果在语法规则中指定抽象语法树的规则，在生成语法分析器的同时，Antlr 还能够生成抽象语法树；最终使用树分析器遍历抽象语法树，完成语义分析和中间代码生成。整个工作在 Anltr 强大的支持下，将变得非常轻松和愉快。

文本处理

当需要文本处理时，首先想到的是正则表达式，使用 Anltr 的词法分析器生成器，可以很容易的完成正则表达式能够完成的所有工作；除此之外使用 Anltr 还可以完成一些正则表达式难以完成的工作，比如识别左括号和右括号的成对匹配等。

基本环境配置

命令方式生成文件

下载antlr

https://www.antlr.org/download/antlr-4.7.2-complete.jar

启动antlr

alias antlr4='java -Xmx500M -cp "/home/deeplearning/antlr/antlr-4.7.2-complete.jar:$CLASSPATH" org.antlr.v4.Tool'

alias grun='java -Xmx500M -cp "/home/deeplearning/antlr/antlr-4.7.2-complete.jar:$CLASSPATH" org.antlr.v4.gui.TestRig'

antlr4

测试案例

定义antlr Hello.g4文件

注意： 逗号必须对齐

grammar Hello; //定义一个名为Hello的语法
r  : 'hello' ID ; // 匹配一个关键字hello和一个紧随其后的标识符
ID : [a-z]+ ;    // 匹配小写字母组成的标识符
WS : [ \t\r\n]+ -> skip; // 忽略空格 Tab 换行

运行生成对应的文件和代码

antlr4 Hello.g4
javac Hello*.java

运行生成树图

grun Hello r -gui

IDEA中antlr4环境配置

参考文章

https://www.baeldung.com/java-antlr

在maven 项目src/main下创建一个目录

mkdir -p antlr4/cn/xh/parse

编写MyParse.g4文件

grammar MyParse;

//parser
prog:stat
;
stat:expr|NEWLINE
;

expr:multExpr(('+'|'-')multExpr)*
;
multExpr:atom(('*'|'/')atom)*
;
atom:'('expr')'
    |INT
    |ID
;

//lexer
ID:('a'..'z'|'A'..'Z')+;
INT:'0'..'9'+;
NEWLINE:'\r'?'\n';
WS:(' '|'\t'|'\n'|'\r')+{skip();};

在pom.xml文件中添加如下内容

<antlr4.version>4.7.2</antlr4.version>
<dependency>
   <groupId>org.antlr</groupId>
   <artifactId>antlr4-runtime</artifactId>
   <version>${antlr4.version}</version>
</dependency>

 <build>
        <plugins>
            <plugin>
                <groupId>org.antlr</groupId>
                <artifactId>antlr4-maven-plugin</artifactId>
                <version>4.7.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>antlr4</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
 </build>

输入maven命令打包

1	mvn package

结果

在target/generated-sources/antlr4/cn/xh/parse 中产生对应的文件

antlr基本语法详解

参考《antlr4权威指南》

定义文件头

使用Antlr 的语法规则来定义算术表达式文法，文件头部是 grammar 关键字，定义文法的名字，必须与文法文件文件的名字相同。

1	grammar MyParse;

自定义一个算术表达式

表达式规则：

算法的优先级需要通过文法规则的嵌套定义来体现，加减法的优先级低于乘除法，表达式 expr 的定义由乘除法表达式 multExpr 和加减法算符 (‘+’|’-‘) 构成；同理，括号的优先级高于乘除法，乘除法表达式 multExpr 通过原子操作数 atom 和乘除法算符 (‘*’|’/’) 构成。

程序有一个语句构成，语句有表达式或者换行符构成。

prog: stat 
; 
stat: expr 
  |NEWLINE 
;

在 Antlr 中语法定义和词法定义通过规则的第一个字符来区别，规定语法定义符号的第一个字母小写，而词法定义符号的第一个字母大写。算术表达式中用到了 4 类记号 ( 在 Antlr 中被称为 Token)，分别是标识符 ID，表示一个变量；常量 INT，表示一个常数；换行符 NEWLINE 和空格 WS，空格字符在语言处理时将被跳过，skip() 是词法分析器类的一个方法

ID:('a'..'z'|'A'..'Z')+;
INT:'0'..'9'+;
NEWLINE:'\r'?'\n';
WS:(' '|'\t'|'\n'|'\r')+{skip();};

案例

antlr文件编写

MyParse.g4文件

grammar MyParse;

//parser
prog:stat
;
stat:expr|NEWLINE
;

expr:multExpr(('+'|'-')multExpr)*
;
multExpr:atom(('*'|'/')atom)*
;
atom:'('expr')'
    |INT
    |ID
;

//lexer
ID:('a'..'z'|'A'..'Z')+;
INT:'0'..'9'+;
NEWLINE:'\r'?'\n';
WS:(' '|'\t'|'\n'|'\r')+{skip();};

使用java调用分析器

参考spark源码中ParseDriver.scala文件中AbstractSqlParser类parse方法

package com.xh.antlr4;

import cn.xh.parse.MyParseLexer;
import cn.xh.parse.MyParseParser;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.apache.spark.sql.catalyst.parser.UpperCaseCharStream;

public class UserMyParse {

    public static void run(String expr) throws Exception {

        //对每一个输入的字符串，构造一个 CodePointCharStream 流 in
        UpperCaseCharStream in = new UpperCaseCharStream(CharStreams.fromString(expr));
        //用 in 构造词法分析器 lexer，词法分析的作用是产生记号
        MyParseLexer lexer = new MyParseLexer(in);

        //用词法分析器 lexer 构造一个记号流 tokens
        CommonTokenStream tokens = new CommonTokenStream(lexer);

        //再使用 tokens 构造语法分析器 parser,至此已经完成词法分析和语法分析的准备工作
        MyParseParser parser = new MyParseParser(tokens);

        //最终调用语法分析器的规则 prog，完成对表达式的验证
        parser.prog();
    }

    public static void main(String[] args) throws Exception {

        String[] testStr = {
                "2",
                "a+b+3",
                "(a-b)+3",
                "a+(b*3)"
        };

        for (String s : testStr) {
            System.out.println("Input expr:" + s);
            run(s);
        }
    }
}

spark sql窗口函数实战

发表于 2019-04-15 | 分类于 sparksql |

| 字数统计: 3,501 | 阅读时长 ≈ 19

参考文章

主要以一些官方文档为参考。
https://cwiki.apache.org/confluence/display/Hive/LanguageManual+WindowingAndAnalytics

https://help.aliyun.com/document_detail/34994.html?spm=a2c4g.11174283.6.650.6f02590e0d209m#h2-url-1

https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html

https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-functions-windows.html

http://xinhstechblog.blogspot.com/2016/04/spark-window-functions-for-dataframes.html

http://cdn2.hubspot.net/hubfs/438089/notebooks/eBook/Introducing_Window_Functions_in_Spark_SQL_Notebook.html

http://blog.madhukaraphatak.com/introduction-to-spark-two-part-5/

https://www.cnblogs.com/piaolingzxh/p/5538783.html

准备数据

object WindowFunctionTest extends BaseSparkSession {
  def main(args: Array[String]): Unit = {
       val sparkConf = new SparkConf()
      .setAppName("WindowFunctionTest")
      .set("spark.master", "local[*]")
    val spark = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()

    import org.apache.spark.sql.expressions.Window
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val df = List(
      ("浙江", "2018-01-01", 500),
      ("浙江", "2018-01-02", 450),
      ("浙江", "2018-01-03", 550),
      ("湖北", "2018-01-01", 250),
      ("湖北", "2018-01-02", 290),
      ("湖北", "2018-01-03", 270)
    ).toDF("site", "date", "user_cnt")
  }
}

平均移动值

DataFrame API方式实现

方式一：

1
2
3

// 窗口定义从 -1(前一行)到 1(后一行)，每一个滑动的窗口总用有3行
 val movinAvgSpec = Window.partitionBy("site").orderBy("date").rowsBetween(-1, 1)
 df.withColumn("MovingAvg", avg(df("user_cnt")).over(movinAvgSpec)).show()

方式二：

 val movinAvgSpec = Window.partitionBy("site").orderBy("date").rowsBetween(-1, 1)
df.select(
  $"site",
  $"date",
  $"amount",
  avg($"user_cnt").over(movinAvgSpec).as("moving_avg_user_cnt")
).show()

sql方式实现

df.createOrReplaceTempView("site_info")
spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       avg(user_cnt) over(partition by site order by date rows between 1 preceding and 1 following) as moving_avg
    |from   site_info
  """.stripMargin).show()

lag函数

说明：取当前记录的前x条数据的指定列，如果没有返回null，有就返回真实值。

DataFrame API方式实现

方式一：

1 2	val lagwSpec = Window.partitionBy("site").orderBy("date") df.withColumn("prevUserCnt", lag(df("user_cnt"), 1).over(lagwSpec)).show()

方式二：

 val lagwSpec = Window.partitionBy("site").orderBy("date")
df.select(
  $"site",
  $"date",
  $"amount",
  lag($"user_cnt").over(movinAvgSpec).as("lag_user_cnt")
).show()

sql方式实现

df.createOrReplaceTempView("site_info")
spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       lag(user_cnt,1) over(partition by  site order by date asc ) as prevUserCnt
    |from   site_info
  """.stripMargin).show()

lead函数

说明：取当前记录的后x条数据的指定列，如果没有返回null，有就返回真实值。

DataFrame API方式实现

方式一：

1 2	val leadwSpec = Window.partitionBy("site").orderBy("date") df.withColumn("lead_user_cnt", lead(df("user_cnt"), 1).over(leadwSpec)).show()

方式二：

val leadwSpec = Window.partitionBy("site").orderBy("date")
  df.select(
    $"site",
    $"date",
    $"user_cnt",
    lead($"user_cnt", 1).over(leadwSpec).as("lead_user_cnt")
  ).show()

sql方式实现

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       lead(user_cnt,1) over(partition by site order by date asc ) as lead_user_cnt
    |from   site_info
  """.stripMargin).show()

结果

+----+----------+--------+-------------+
|site|      date|user_cnt|lead_user_cnt|
+----+----------+--------+-------------+
|湖北|2018-01-01|     250|          290|
|湖北|2018-01-02|     290|          270|
|湖北|2018-01-03|     270|         null|
|浙江|2018-01-01|     500|          450|
|浙江|2018-01-02|     450|          550|
|浙江|2018-01-03|     550|         null|
+----+----------+--------+-------------+

FIRST_VALUE函数

说明：该函数用于获取分组排序后最第一条记录的字段值。

DataFrame API方式实现

方式一：

1 2	val firstValuewSpec = Window.partitionBy("site").orderBy("date") df.withColumn("first_value_user_cnt", first("user_cnt").over(firstValuewSpec)).show()

方式二：

val firstValuewSpec = Window.partitionBy("site").orderBy("date")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  first($"user_cnt").over(firstValuewSpec).as("first_value_user_cnt")).show()

sql方式实现

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       first_value(user_cnt) over(partition by site order by date asc ) as first_value_user_cnt
    |from   site_info
  """.stripMargin).show()

结果

+----+----------+--------+--------------------+
|site|      date|user_cnt|first_value_user_cnt|
+----+----------+--------+--------------------+
|湖北|2018-01-01|     250|                 250|
|湖北|2018-01-02|     290|                 250|
|湖北|2018-01-03|     270|                 250|
|浙江|2018-01-01|     500|                 500|
|浙江|2018-01-02|     450|                 500|
|浙江|2018-01-03|     550|                 500|
+----+----------+--------+--------------------+

LAST_VALUE函数

说明：该函数用于获取分组排序后最后一条记录的字段值。

DataFrame API方式实现

方式一：

1 2	val lastValuewSpec = Window.partitionBy("site").orderBy("date").rowsBetween(Long.MinValue, Long.MaxValue) df.withColumn("last_value_user_cnt", last("user_cnt").over(lastValuewSpec)).show()

方式二：

val lastValuewSpec = Window.partitionBy("site").orderBy("date").rowsBetween(Long.MinValue, Long.MaxValue)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  last($"user_cnt").over(lastValuewSpec).as("last_value_user_cnt")).show()

sql方式实现

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       last_value(user_cnt) over(partition by site order by date asc rows between unbounded preceding and unbounded following ) as last_value_user_cnt
    |from   site_info
  """.stripMargin).show()

结果

+----+----------+--------+-------------------+
|site|      date|user_cnt|last_value_user_cnt|
+----+----------+--------+-------------------+
|湖北|2018-01-01|     250|                270|
|湖北|2018-01-02|     290|                270|
|湖北|2018-01-03|     270|                270|
|浙江|2018-01-01|     500|                550|
|浙江|2018-01-02|     450|                550|
|浙江|2018-01-03|     550|                550|
+----+----------+--------+-------------------+

COUNT

说明：该函数用于计算计数值。

不指定order by

###
方式一：

1 2	val counWSpec = Window.partitionBy("site") df.withColumn("count", count("user_cnt").over(counWSpec)).show()

方式二：

val counWSpec = Window.partitionBy("site")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  count($"user_cnt").over(counWSpec).as("count")).show()

sql方式实现

spark.sql(
     """
       | select  site,
       |         date,
       |         user_cnt,
       |         count(user_cnt) over(partition by site) as count
       |from     site_info
     """.stripMargin).show()

结果

+----+----------+--------+-----+
|site|      date|user_cnt|count|
+----+----------+--------+-----+
|湖北|2018-01-01|     250|    3|
|湖北|2018-01-02|     290|    3|
|湖北|2018-01-03|     270|    3|
|浙江|2018-01-01|     500|    3|
|浙江|2018-01-02|     450|    3|
|浙江|2018-01-03|     550|    3|
+----+----------+--------+-----+

指定order by

指定order by时，返回当前窗口内从开始行到当前行的累计计数值。

DataFrame API方式实现

方式一：

1 2	val counWSpec = Window.partitionBy("site").orderBy('date.asc) df.withColumn("count", count("user_cnt").over(counWSpec)).show()

方式二：

val counWSpec = Window.partitionBy("site").orderBy("date")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  count($"user_cnt").over(counWSpec).as("count"))

sql方式实现

spark.sql(
  """
    | select  site,
    |         date,
    |         user_cnt,
    |         count(user_cnt) over(partition by site order by date) as count
    |from     site_info
  """.stripMargin).show()

结果

+----+----------+--------+-----+
|site|      date|user_cnt|count|
+----+----------+--------+-----+
|湖北|2018-01-01|     250|    1|
|湖北|2018-01-02|     290|    2|
|湖北|2018-01-03|     270|    3|
|浙江|2018-01-01|     500|    1|
|浙江|2018-01-02|     450|    2|
|浙江|2018-01-03|     550|    3|
+----+----------+--------+-----+

sum函数

说明：该函数用于计算汇总值。

不指定order by

DataFrame API方式实现

方式一：

1 2	val sumWSpec = Window.partitionBy("site") df.withColumn("sum_user_cnt", sum("user_cnt").over(sumWSpec)).show()

方式二：

val sumWSpec = Window.partitionBy("site")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  sum($"user_cnt").over(sumWSpec).as("sum_user_cnt")
).show()

sql方式实现

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       sum(user_cnt) over(partition by site ) as sum_user_cnt
    |from   site_info
  """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|sum_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|         810|
|湖北|2018-01-02|     290|         810|
|湖北|2018-01-03|     270|         810|
|浙江|2018-01-01|     500|        1500|
|浙江|2018-01-02|     450|        1500|
|浙江|2018-01-03|     550|        1500|
+----+----------+--------+------------+

指定order by

DataFrame API方式实现

方式一：

1 2	val sumWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue) df.withColumn("sum_user_cnt", sum("user_cnt").over(sumWSpec))

方式二：

val sumWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  sum($"user_cnt").over(sumWSpec).as("sum_user_cnt")
).show()

sql方式实现

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       sum(user_cnt) over(partition by site order by date asc  rows between unbounded preceding and unbounded following ) as sum_user_cnt
    |from   site_info
  """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|sum_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|         810|
|湖北|2018-01-02|     290|         810|
|湖北|2018-01-03|     270|         810|
|浙江|2018-01-01|     500|        1500|
|浙江|2018-01-02|     450|        1500|
|浙江|2018-01-03|     550|        1500|
+----+----------+--------+------------+

min函数

不指定order by

DataFrame API方式实现

方式一：

1 2	val minWSpec = Window.partitionBy("site") df.withColumn("min_user_cnt", min("user_cnt").over(minWSpec)).show()

方式二：

df.select(
  $"site",
  $"date",
  $"user_cnt",
  min($"user_cnt").over(minWSpec)
).show()

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        min(user_cnt) over(partition by site) as min_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+----------------------------------------------------------+
|site|      date|user_cnt|min(user_cnt) OVER (PARTITION BY site unspecifiedframe$())|
+----+----------+--------+----------------------------------------------------------+
|湖北|2018-01-01|     250|                                                       250|
|湖北|2018-01-02|     290|                                                       250|
|湖北|2018-01-03|     270|                                                       250|
|浙江|2018-01-01|     500|                                                       450|
|浙江|2018-01-02|     450|                                                       450|
|浙江|2018-01-03|     550|                                                       450|
+----+----------+--------+----------------------------------------------------------+

指定order by

方式一：

1 2	val minWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue) df.withColumn("min_user_cnt", min("user_cnt").over(minWSpec)).show()

方式二：

val minWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  min($"user_cnt").over(minWSpec)
)

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        min(user_cnt) over(partition by site order by date asc rows between unbounded preceding and unbounded following) as min_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+----------------------------------------------------------+
|site|      date|user_cnt|min(user_cnt) OVER (PARTITION BY site unspecifiedframe$())|
+----+----------+--------+----------------------------------------------------------+
|湖北|2018-01-01|     250|                                                       250|
|湖北|2018-01-02|     290|                                                       250|
|湖北|2018-01-03|     270|                                                       250|
|浙江|2018-01-01|     500|                                                       450|
|浙江|2018-01-02|     450|                                                       450|
|浙江|2018-01-03|     550|                                                       450|
+----+----------+--------+----------------------------------------------------------+

max函数

不指定order by

DataFrame API方式实现

方式一：

1 2	val maxWSpec = Window.partitionBy("site") df.withColumn("min_user_cnt", max("user_cnt").over(maxWSpec))

方式二：

val maxWSpec = Window.partitionBy("site")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  max($"user_cnt").over(maxWSpec).as("max_user_cnt")
)

sql方式实现

spark.sql(
   """
     |select  site,
     |        date,
     |        user_cnt,
     |        max(user_cnt) over(partition by site ) as min_user_cnt
     |from    site_info
   """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|min_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|         290|
|湖北|2018-01-02|     290|         290|
|湖北|2018-01-03|     270|         290|
|浙江|2018-01-01|     500|         550|
|浙江|2018-01-02|     450|         550|
|浙江|2018-01-03|     550|         550|
+----+----------+--------+------------+

指定order by

DataFrame API方式实现

方式一：

1 2	val maxWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue) df.withColumn("min_user_cnt", max("user_cnt").over(maxWSpec)).show()

方式二：

val maxWSpec = Window.partitionBy("site").orderBy('date asc).rowsBetween(Long.MinValue, Long.MaxValue)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  max($"user_cnt").over(maxWSpec).as("max_user_cnt")
).show()

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        max(user_cnt) over(partition by site order by date asc rows between unbounded preceding and unbounded following) as min_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|min_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|         290|
|湖北|2018-01-02|     290|         290|
|湖北|2018-01-03|     270|         290|
|浙江|2018-01-01|     500|         550|
|浙江|2018-01-02|     450|         550|
|浙江|2018-01-03|     550|         550|
+----+----------+--------+------------+

avg函数

说明：该函数用于计算平均值。

不指定order by

DataFrame API方式实现

方式一：

1 2	val avgWSpec = Window.partitionBy("site") df.withColumn("avg_user_cnt", avg("user_cnt").over(avgWSpec)).show()

方式二：

val avgWSpec = Window.partitionBy("site")
df.select(
  $"site",
  $"date",
  $"user_cnt",
  avg("user_cnt").over(avgWSpec).as("avg_user_cnt")
).show()

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        avg(user_cnt) over(partition by site ) as avg_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|avg_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|       270.0|
|湖北|2018-01-02|     290|       270.0|
|湖北|2018-01-03|     270|       270.0|
|浙江|2018-01-01|     500|       500.0|
|浙江|2018-01-02|     450|       500.0|
|浙江|2018-01-03|     550|       500.0|
+----+----------+--------+------------+

指定order by

DataFrame API方式实现

方式一：

1 2	val avgWSpec = Window.partitionBy("site").orderBy("date").rowsBetween(Long.MinValue, Long.MaxValue) df.withColumn("avg_user_cnt", avg("user_cnt").over(avgWSpec)).show()

方式二：

val avgWSpec = Window.partitionBy("site").orderBy("date").rowsBetween(Long.MinValue, Long.MaxValue)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  avg("user_cnt").over(avgWSpec).as("avg_user_cnt")
)

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        avg(user_cnt) over(partition by site order by date  asc rows between unbounded preceding and unbounded following ) as avg_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+------------+
|site|      date|user_cnt|avg_user_cnt|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|       270.0|
|湖北|2018-01-02|     290|       270.0|
|湖北|2018-01-03|     270|       270.0|
|浙江|2018-01-01|     500|       500.0|
|浙江|2018-01-02|     450|       500.0|
|浙江|2018-01-03|     550|       500.0|
+----+----------+--------+------------+

rank函数

说明：该函数用于计算排名。

DataFrame API方式实现

方式一：

1 2	val rankWSpec = Window.partitionBy("site").orderBy('user_cnt.desc) df.withColumn("rank", rank().over(rankWSpec))

方式二：

val rankWSpec = Window.partitionBy("site").orderBy('user_cnt.desc)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  rank().over(rankWSpec).as("rank")
).show()

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        rank() over(partition by site order by user_cnt desc) as rank_user_cnt
    |from    site_info
  """.stripMargin).show()

结果

1
2

row_number over 函数

DataFrame API方式实现

方式一：

1 2	val rowNUmberWSpec = Window.partitionBy("site").orderBy('date desc, 'user_cnt desc) df.withColumn("row_num", row_number().over(rowNUmberWSpec)).show()

方式二：

val rowNUmberWSpec = Window.partitionBy("site").orderBy('date desc, 'user_cnt desc)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  row_number().over(rowNUmberWSpec).as("row_num")
).show()

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        row_number() over(partition by site order by date desc , user_cnt desc ) as row_num
    |from    site_info
    |
  """.stripMargin)

结果

+----+----------+--------+-------+
|site|      date|user_cnt|row_num|
+----+----------+--------+-------+
|湖北|2018-01-03|     270|      1|
|湖北|2018-01-02|     290|      2|
|湖北|2018-01-01|     250|      3|
|浙江|2018-01-03|     550|      1|
|浙江|2018-01-02|     450|      2|
|浙江|2018-01-01|     500|      3|
+----+----------+--------+-------+

dense_rank函数

说明：该函数用于计算连续排名。

DataFrame API方式实现

方式一：

1 2	val denseRankWSpec = Window.partitionBy("site").orderBy('date asc) df.withColumn("dense_rank", dense_rank() over (denseRankWSpec)).show()

方式二：

val denseRankWSpec = Window.partitionBy("site").orderBy('date asc)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  dense_rank().over(denseRankWSpec).as("dense_rank")
).show()

sql方式实现

  spark.sql(
    """
      |select site,
      |       date,
      |       user_cnt,
      |       dense_rank() over(partition by site order by date asc ) as dense_rank
      |from   site_info
      |
""".stripMargin).show()

结果

+----+----------+--------+----------+
|site|      date|user_cnt|dense_rank|
+----+----------+--------+----------+
|湖北|2018-01-01|     250|         1|
|湖北|2018-01-02|     290|         2|
|湖北|2018-01-03|     270|         3|
|浙江|2018-01-01|     500|         1|
|浙江|2018-01-02|     450|         2|
|浙江|2018-01-03|     550|         3|
+----+----------+--------+----------+

percent_rank函数

说明：该函数用于计算一组数据中某行的相对排名。

DataFrame API方式实现

方式一：

1 2	val percentRankWSpec = Window.partitionBy("site").orderBy('date asc) df.withColumn("percent_rank", percent_rank() over (percentRankWSpec)).show()

方式二：

val percentRankWSpec = Window.partitionBy("site").orderBy('date asc)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  percent_rank().over(percentRankWSpec).as("percent_rank")
).show()

sql方式实现

  spark.sql(
    """
      |select site,
      |       date,
      |       user_cnt,
      |       percent_rank() over(partition by site order by date asc ) as percent_rank
      |from   site_info
      |
""".stripMargin).show()

结果

|site|      date|user_cnt|percent_rank|
+----+----------+--------+------------+
|湖北|2018-01-01|     250|         0.0|
|湖北|2018-01-02|     290|         0.5|
|湖北|2018-01-03|     270|         1.0|
|浙江|2018-01-01|     500|         0.0|
|浙江|2018-01-02|     450|         0.5|
|浙江|2018-01-03|     550|         1.0|
+----+----------+--------+------------+

ntile函数

说明：用于将分组数据按照顺序切分成n片，并返回当前切片值，如果切片不均匀，默认增加第一个切片的分布。

DataFrame API方式实现

方式一：

1 2	val ntileRankWSpec = Window.partitionBy("site").orderBy('date asc) df.withColumn("ntile", ntile(2).over(ntileRankWSpec)).show()

方式二：

val ntileRankWSpec = Window.partitionBy("site").orderBy('date asc)
df.select(
  $"site",
  $"date",
  $"user_cnt",
  ntile(2).over(ntileRankWSpec).as("ntile")
)

sql方式实现

spark.sql(
  """
    |select  site,
    |        date,
    |        user_cnt,
    |        ntile(2) over(partition by site order by date) as ntile
    |from    site_info
  """.stripMargin).show()

结果

+----+----------+--------+-----+
|site|      date|user_cnt|ntile|
+----+----------+--------+-----+
|湖北|2018-01-01|     250|    1|
|湖北|2018-01-02|     290|    1|
|湖北|2018-01-03|     270|    2|
|浙江|2018-01-01|     500|    1|
|浙江|2018-01-02|     450|    1|
|浙江|2018-01-03|     550|    2|
+----+----------+--------+-----+

致谢！

本人能力有限，博客错误难免，有错往将错误发送到邮箱(t_spider@aliyun.com)

sparksql实战案例

发表于 2019-04-10 | 分类于 sparksql |

| 字数统计: 2,053 | 阅读时长 ≈ 12

累计统计

准备数据

access.csv

A,2015-01,5
A,2015-01,15
A,2015-01,5
A,2015-01,8
A,2015-02,4
A,2015-02,6
A,2015-03,16
A,2015-03,22
A,2015-04,10
A,2015-04,50
B,2015-01,5
B,2015-01,25
B,2015-02,10
B,2015-02,5
B,2015-03,23
B,2015-03,10
B,2015-03,1
B,2015-04,10
B,2015-04,50

准备环境

object AccumulatorCount {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("DateFrameFromJsonScala")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    import org.apache.spark.sql.expressions.Window
    import org.apache.spark.sql.functions._
    import spark.implicits._
    // 读取数据
    val usersDF = spark.read.format("csv")
      .option("sep", ",")
      .option("inferSchema", "true")
      .option("header", "false")
      .load("src/main/resources/access.csv")
      .toDF("name", "mounth", "amount")
  }
}

具体实现逻辑

DataFrame API 方式

方式一:

1
2
3

// rowsBetween(Long.MinValue, 0):窗口的大小是按照排序从最小值到当前行
val accuCntSpec = Window.partitionBy("name").orderBy("mounth").rowsBetween(Long.MinValue, 0)
usersDF.withColumn("acc_amount", sum(usersDF("amount")).over(accuCntSpec)).show()

方式二

usersDF.select(
  $"name",
  $"mounth",
  $"amount",
  sum($"amount").over(accuCntSpec).as("acc_amount")
).show()

sql方式

思路：根据DF算子意思，找到SqlBase.g4文件，看看是否有该类sql支持。
在SqlBase.g4文件中刚好找到如下内容

windowFrame
    : frameType=RANGE start=frameBound
    | frameType=ROWS start=frameBound
    | frameType=RANGE BETWEEN start=frameBound AND end=frameBound
    | frameType=ROWS BETWEEN start=frameBound AND end=frameBound
    ;

frameBound
    : UNBOUNDED boundType=(PRECEDING | FOLLOWING)
    | boundType=CURRENT ROW
    | expression boundType=(PRECEDING | FOLLOWING)
    ;

在spark源码sql模块core项目org.apache.spark.sql.execution包中找到SQLWindowFunctionSuite类找到如下测试方法

test("window function: multiple window expressions in a single expression") {
   val nums = sparkContext.parallelize(1 to 10).map(x => (x, x % 2)).toDF("x", "y")
   nums.createOrReplaceTempView("nums")

   val expected =
     Row(1, 1, 1, 55, 1, 57) ::
       Row(0, 2, 3, 55, 2, 60) ::
       Row(1, 3, 6, 55, 4, 65) ::
       Row(0, 4, 10, 55, 6, 71) ::
       Row(1, 5, 15, 55, 9, 79) ::
       Row(0, 6, 21, 55, 12, 88) ::
       Row(1, 7, 28, 55, 16, 99) ::
       Row(0, 8, 36, 55, 20, 111) ::
       Row(1, 9, 45, 55, 25, 125) ::
       Row(0, 10, 55, 55, 30, 140) :: Nil

   val actual = sql(
     """
       |SELECT
       |  y,
       |  x,
       |  sum(x) OVER w1 AS running_sum,
       |  sum(x) OVER w2 AS total_sum,
       |  sum(x) OVER w3 AS running_sum_per_y,
       |  ((sum(x) OVER w1) + (sum(x) OVER w2) + (sum(x) OVER w3)) as combined2
       |FROM nums
       |WINDOW w1 AS (ORDER BY x ROWS BETWEEN UnBOUNDED PRECEDiNG AND CuRRENT RoW),
       |       w2 AS (ORDER BY x ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOuNDED FoLLOWING),
       |       w3 AS (PARTITION BY y ORDER BY x ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
     """.stripMargin)

   checkAnswer(actual, expected)

   spark.catalog.dropTempView("nums")
 }

下面就可以开心的照着案例写sql去了，真嗨皮！！！！

usersDF.createOrReplaceTempView("access")
spark.sql(
  """
    |select name,
    |       mounth,
    |       amount,
    |	      sum(amount) over (partition by name order by mounth asc  rows between unbounded preceding and current row ) as acc_amount
    |from   access
    |
  """.stripMargin).show()

累加N天之前,假设N=3

DataFrame API方式

val preThreeAccuCntSpec = Window.partitionBy("name").orderBy("mounth").rowsBetween(-3, 0)
usersDF.select(
  $"name",
  $"mounth",
  $"amount",
  sum($"amount").over(preThreeAccuCntSpec).as("acc_amount")).show()

sql方式

spark.sql(
  """
    |select name,
    |       mounth,
    |       amount,
    |	      sum(amount) over (partition by name order by mounth asc rows between 3 preceding and current row) as acc_amount
    |from   access
    |
  """.stripMargin).show()

累加前3天，后3天

API方式

val preThreeFiveAccuCntSpec = Window.partitionBy("name").orderBy("mounth").rowsBetween(3, 3)
usersDF.select(
  $"name",
  $"mounth",
  $"amount",
  sum($"amount").over(preThreeFiveAccuCntSpec).as("acc_amount"))

sql方式

spark.sql(
   """
     |select name,
     |       mounth,
     |       amount,
     |	      sum(amount) over (partition by name order by mounth asc rows between 3 preceding and 3 following) as acc_amount
     |from   access
     |
   """.stripMargin).show()

基本窗口函数案例

准备环境

object WindowFunctionTest extends BaseSparkSession {
  def main(args: Array[String]): Unit = {
   val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("WindowFunctionTest")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    import org.apache.spark.sql.expressions.Window
    import org.apache.spark.sql.functions._
    import spark.implicits._
    import org.apache.spark.sql.expressions.Window
    import org.apache.spark.sql.functions._
    import spark.implicits._
    val df = List(
      ("位置1", "2018-01-01", 50),
      ("位置1", "2018-01-02", 45),
      ("位置1", "2018-01-03", 55),
      ("位置2", "2018-01-01", 25),
      ("位置2", "2018-01-02", 29),
      ("位置2", "2018-01-03", 27)
    ).toDF("site", "date", "user_cnt")
  }
}

平均移动值

DataFrame API方式实现

1
2
3

// 窗口定义从 -1(前一行)到 1(后一行)	，每一个滑动的窗口总用有3行
 val movinAvgSpec = Window.partitionBy("site").orderBy("date").rowsBetween(-1, 1)
    df.withColumn("MovingAvg", avg(df("user_cnt")).over(movinAvgSpec)).show()

sql方式实现

df.createOrReplaceTempView("site_info")
spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       avg(user_cnt) over(partition by site order by date rows between 1 preceding and 1 following) as moving_avg
    |from   site_info
  """.stripMargin).show()

前一行数据

DataFrame API方式实现

1 2	val lagwSpec = Window.partitionBy("site").orderBy("date") df.withColumn("prevUserCnt", lag(df("user_cnt"), 1).over(lagwSpec)).show()

sql方式实现

df.createOrReplaceTempView("site_info")
spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       lag(user_cnt,1) over(partition by  site order by date asc ) as prevUserCnt
    |from   site_info
  """.stripMargin).show()

排名

DataFrame API方式实现

1 2	val rankwSpec = Window.partitionBy("site").orderBy("date") df.withColumn("rank", rank().over(rankwSpec))

sql方式

spark.sql(
  """
    |select site,
    |       date,
    |       user_cnt,
    |       rank() over(partition by  site order by date asc ) as prevUserCnt
    |from   site_info
  """.stripMargin).show()

分组topn和分组取最小

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}
object GroupBy {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local")
      .appName("DateFrameFromJsonScala")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()
    val rows = spark.sparkContext.parallelize(
      List(
        ("shop2", "2018-02-22", 1),
        ("shop2", "2018-02-27", 1),
        ("shop2", "2018-03-13", 1),
        ("shop2", "2018-03-20", 5),
        ("shop1", "2018-03-27", 1),
        ("shop1", "2018-04-03", 1),
        ("shop1", "2018-04-10", 1),
        ("shop1", "2018-04-17", 1),
        ("shop2", "2018-04-28", 1),
        ("shop2", "2018-04-05", 10),
        ("shop2", "2018-04-09", 1)))
    val rowRDD = rows.map(t => Row(t._1, t._2, t._3))
    val schema = StructType(
      Array(
        StructField("shop", StringType),
        StructField("ycd_date", StringType),
        StructField("ycd_num", IntegerType)))
    val df = spark.createDataFrame(rowRDD, schema)
    df.createOrReplaceTempView("ycd_order")
    // 分组topN
    val topN = spark.sql("select * from (SELECT o.shop,o.ycd_date, row_number() over (PARTITION BY o.shop ORDER BY o.ycd_date DESC) rank FROM ycd_order as o) o1 where  rank < 2")
    // 根据某一个字段分组,取某一个字段的最小值
    val groupMin = spark.sql("select o.shop,min(o.ycd_num) as min_num from ycd_order as o group by o.shop order by min_num ")
    topN.show()
    spark.stop()
  }
}

优雅方式定义scheme

def getScheme(): StructType = {
    val schemaString = "store_id:String,order_date:String,sale_amount: Int"
    val fields = schemaString.split(",")
      .map(fieldName =>
        StructField(fieldName.split(":")(0).trim,
          fieldName.split(":")(1).trim match {
            case "String" => StringType
            case "Int" => IntegerType
          }, true))
    StructType(fields)
  }

保存小数点后n位

10表示总的位数，2表示保留几位小数，10要>=实际的位数，否则为NULL

1	spark.sql("select cast(sale_amount as decimal(10, 2))from ycd").show()

重命名行

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{SQLContext, Row, DataFrame, Column}
import org.apache.spark.ml.feature.VectorAssembler
val firstDF = spark.createDataFrame(Seq(
  (1, 1, 2, 3, 8, 4, 5),
  (2, 4, 3, 8, 7, 9, 8),
  (3, 6, 1, 9, 2, 3, 6),
  (4, 7, 8, 6, 9, 4, 5),
  (5, 9, 2, 7, 8, 7, 3),
  (6, 1, 1, 4, 2, 8, 4)
)).toDF()
val colNames = Seq("uid", "col1", "col2", "col3", "col4", "col5", "col6")
val secondDF = firstDF.toDF(colNames: _*)

转置

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{array, col, explode, lit, struct}

val df = spark.createDataFrame(Seq(
  (1, 1, 2, 3, 8, 4, 5),
  (2, 4, 3, 8, 7, 9, 8),
  (3, 6, 1, 9, 2, 3, 6),
  (4, 7, 8, 6, 9, 4, 5),
  (5, 9, 2, 7, 8, 7, 3),
  (6, 1, 1, 4, 2, 8, 4)
)).toDF("uid", "col1", "col2", "col3", "col4", "col5", "col6")

df.show(10,false)

// Create the transpose user defined function.
// Imputs:
//   transDF: The dataframe which will be transposed
//   transBy: The column that the dataframe will be transposed by
// Outputs:
//   Dataframe datatype consisting of three columns:
//     transBy
//     column_name
//     column_value
def transposeUDF(transDF: DataFrame, transBy: Seq[String]): DataFrame = {
  val (cols, types) = transDF.dtypes.filter{ case (c, _) => !transBy.contains(c)}.unzip
  require(types.distinct.size == 1)      

  val kvs = explode(array(
    cols.map(c => struct(lit(c).alias("column_name"), col(c).alias("column_value"))): _*
  ))
  val byExprs = transBy.map(col(_))

  transDF
    .select(byExprs :+ kvs.alias("_kvs"): _*)
    .select(byExprs ++ Seq($"_kvs.column_name", $"_kvs.column_value"): _*)
}
transposeUDF(df, Seq("uid")).show(12,false)
Output:
df.show(10,false)
+---+----+----+----+----+----+----+
|uid|col1|col2|col3|col4|col5|col6|
+---+----+----+----+----+----+----+
|1  |1   |2   |3   |8   |4   |5   |
|2  |4   |3   |8   |7   |9   |8   |
|3  |6   |1   |9   |2   |3   |6   |
|4  |7   |8   |6   |9   |4   |5   |
|5  |9   |2   |7   |8   |7   |3   |
|6  |1   |1   |4   |2   |8   |4   |
+---+----+----+----+----+----+----+

transposeUDF(df, Seq("uid")).show(12,false)
+---+-----------+------------+
|uid|column_name|column_value|
+---+-----------+------------+
|1  |col1       |1           |
|1  |col2       |2           |
|1  |col3       |3           |
|1  |col4       |8           |
|1  |col5       |4           |
|1  |col6       |5           |
|2  |col1       |4           |
|2  |col2       |3           |
|2  |col3       |8           |
|2  |col4       |7           |
|2  |col5       |9           |
|2  |col6       |8           |
+---+-----------+------------+
only showing top 12 rows

DynamicVariable详解

发表于 2019-02-13 | 分类于 scala |

| 字数统计: 304 | 阅读时长 ≈ 1

内容来源于：https://stackoverflow.com/questions/5116352/when-we-should-use-scala-util-dynamicvariable

说明

1 2	protected val _response = new DynamicVariable[HttpServletResponse](null) protected val _request = new DynamicVariable[HttpServletRequest](null)

DynamicVariable是贷款和动态范围模式的实现。 DynamicVariable的使用情况与Java中的ThreadLocal非常相似(事实上，DynamicVariable在后台使用InheritableThreadLocal) – 当需要在一个封闭的作用域内进行计算时，每个线程都有自己的副本的变量值：

1
2
3

dynamicVariable.withValue(value){ valueInContext =>
  // value used in the context
}

由于DynamicVariable使用可继承的ThreadLocal，变量的值被传递给上下文中生成的线程：

dynamicVariable.withValue(value){ valueInContext =>
  spawn{
    // value is passed to the spawned thread
  }
}

DynamicVariable(和ThreadLocal)在Scalatra中使用的原因与在许多其他框架(Lift，Spring，Struts等)中使用的相同 – 它是一种非侵入性的方式来存储和传递上下文(线程)特定的信息。

使HttpServletResponse和HttpServletRequest动态变量(并因此，绑定到处理请求的特定线程)只是最简单的方法来获取它们在代码中的任何地方(不通过方法参数或任何其他显式)。

案例

val dyn = new DynamicVariable[String]("withoutValue")
def print=println(dyn.value)
print
dyn.withValue("withValue") {
  print
}
print

结果
withoutValue
withValue
withoutValue

java concurrent包类详解

发表于 2019-02-13 | 分类于 java |

| 字数统计: 549 | 阅读时长 ≈ 2

Semaphore

原文：https://zhuanlan.zhihu.com/p/27314456
Semaphore(信号量)是java.util.concurrent下的一个工具类.用来控制可同时访问特定资源的线程数.内部是通过维护父类(AQS)的 int state值实现.
Semaphore中有一个”许可”的概念:

访问特定资源前，先使用acquire(1)获得许可，如果许可数量为0，该线程则一直阻塞，直到有可用许可。
访问资源后，使用release()释放许可。

这个许可在构造时传入,赋给state值,它等同于state.

Semaphore应用场景
系统中某类资源比较紧张,只能被有限的线程访问,此时适合使用信号量。
Semaphore用来控制访问某资源的线程数,比如数据库连接.假设有这个的需求，读取几万个文件的数据到数据库中，由于文件读取是IO密集型任务，可以启动几十个线程并发读取，但是数据库连接数只有20个，这时就必须控制最多只有20个线程能够拿到数据库连接进行操作。这个时候，就可以使用Semaphore做流量控制。
使用案例：
spark LiveListenerBus类

private val eventLock = new Semaphore(0)

private val listenerThread = new Thread(name) {
  setDaemon(true)
  override def run(): Unit = Utils.tryOrStopSparkContext(sparkContext) {
    LiveListenerBus.withinListenerThread.withValue(true) {
      while (true) {
        eventLock.acquire()
        self.synchronized {
          processingEvent = true
        }
        try {
          // 消费消息
          val event = eventQueue.poll
          if (event == null) {
            // Get out of the while loop and shutdown the daemon thread
            if (!stopped.get) {
              throw new IllegalStateException("Polling `null` from eventQueue means" +
                " the listener bus has been stopped. So `stopped` must be true")
            }
            return
          }
          postToAll(event)
        } finally {
          self.synchronized {
            processingEvent = false
          }
        }
      }
    }
  }
}

Semaphore属于一种较常见的限流手段，Google Guava封装了一层。

//JDK API：流速控制在每秒执行100个任务
final Semaphore semaphore = new Semaphore(100);
void submitTasks(List<Runnable> tasks, Executor executor) {
    for (Runnable task : tasks) {
        semaphore.acquire(); // 也许需要等待
        executor.execute(task);
        semaphore.release();
    }
}

//Google Guava API：流速控制在每秒执行100个任务
final RateLimiter rateLimiter = RateLimiter.create(100);
void submitTasks(List<Runnable> tasks, Executor executor) {
    for (Runnable task : tasks) {
        rateLimiter.acquire(); // 也许需要等待
        executor.execute(task);
    }
}

scala枚举

发表于 2019-01-09 | 分类于 scala |

| 字数统计: 175 | 阅读时长 ≈ 1

定义枚举对象

/** 定义一个枚举类 */
private[deploy] object SparkSubmitAction extends Enumeration {
  // 声明枚举对外暴露的变量类型
  type SparkSubmitAction = Value
  // 枚举的定义
  val SUBMIT, KILL, REQUEST_STATUS = Value
}

使用枚举

/** 定义一个枚举类 */
object SparkSubmitAction extends Enumeration {
  // 声明枚举对外暴露的变量类型
  type SparkSubmitAction = Value
  // 枚举的定义
  val SUBMIT, KILL, REQUEST_STATUS = Value

  def getAction(action: SparkSubmitAction){
  action match {
    case SUBMIT => println ("action is " + action)
    case KILL => println ("action is " + action)
    case REQUEST_STATUS => println ("action is " + action)
    case _ => println ("Unknown type")
  }
}
}

测试用例

object EnumerationTest {
  def main(args: Array[String]): Unit = {
    val action = SparkSubmitAction.apply(1)
    SparkSubmitAction.getAction(action)

    val action1 = SparkSubmitAction.withName("quit")
    SparkSubmitAction.getAction(action1)
  }
}

scala学习笔记

发表于 2019-01-09 | 分类于 scala |

| 字数统计: 6,121 | 阅读时长 ≈ 27

高级函数

高阶函数是指使用其他函数作为参数、或者返回一个函数作为结果的函数。

匿名函数

1	(x: Int) => x * 3

带函数参数的函数

Scala集合类（collections）的高阶函数map。

def map[B, That](f: A => B)(implicit bf: CanBuildFrom[Repr, B, That]): That = {
   def builder = {
     val b = bf(repr)
     b.sizeHint(this)
     b
   }
   val b = builder
   for (x <- this) b += f(x)
   b.result
 }


  val seq = Seq(100, 200, 300)

   def doubleSalary(x: Int): Int = {
     x * 2
   }
   seq.map(doubleSalary).foreach(x => println(x))
   seq.map(x => x * 2)
   seq.map(_ * 2)

闭包

1
2
3

def sum(x: Int) = (y: Int) => x + y
val first = sum(1)
val second = first(4)

嵌套方法

在Scala中可以嵌套定义方法。

def factorial(x: Int): Int = {
    def fact(x: Int, accumulator: Int): Int = {
      if (x <= 1) accumulator
      else fact(x - 1, x * accumulator)
    }  
    fact(x, 1)
 }

 println("Factorial of 2: " + factorial(2))
 println("Factorial of 3: " + factorial(3))

柯里化

柯里化：指将原来接受两个参数的函数变成新的接受一个参数的过程。
接受两个参数的过程：

1	def mul(x:Int , y: Int )= x * y

接受一个参数的过程

1 2	def mul(x: Int) = (y: Int) => x + y mul(1)(2)

scala支持简写成如下的柯里化：

1
2
3

def mul(x: Int)(y: Int) = x + y

mul(1)(2)

在Scala集合中定义的特质TraversableOnce[+A]。
Traversable：能横过的；能越过的；可否定的。

def foldLeft[B](z: B)(op: (B, A) => B): B = {
    var result = z
    this foreach (x => result = op(result, x))
    result
  }
```   
foldLeft从左到右，以此将一个二元运算op应用到初始值z和该迭代器（traversable)的所有元素上。以下是该函数的一个用例：

从初值0开始, 这里 foldLeft 将函数 (m, n) => m + n 依次应用到列表中的每一个元素和之前累积的值上。  
```scala
val numbers = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val res = numbers.foldLeft(0)((m, n) => m + n)
val res2 = numbers.foldLeft(0)(_ + _)
println(res) // 55
pringln(res2)

多参数列表有更复杂的调用语法，因此应该谨慎使用。
建议的使用场景包括:

单一的函数参数。
在某些情况下存在单一的函数参数时，例如上述例子foldLeft中的op，多参数列表可以使得传递匿名函数作为参数的语法更为简洁。如果不使用多参数列表，代码可能像这样：
1
numbers.foldLeft(0, {(m: Int, n: Int) => m + n})

隐式（IMPLICIT）参数。

1	def execute(arg: Int)(implicit ec: ExecutionContext) = ???

模式匹配

match对应java里的Switch,不过它写在选择器表达式之后。

选择器 match {备选项}
取代了：
switch(选择器){备选项}

match与switch的比较

匹配表达式可以被看做java风格switch的泛化。
match的不同：
1、match是scala表达式，也就是说，它始终以值作为结果;
2、 scala的备选项表达式永远不会”掉到”下一个case;
3、如果没有模式匹配，MatchErro异常会被抛出。

提取器对象

提取器对象是一个包含有 unapply 方法的单例对象。apply 方法就像一个构造器，接受参数然后创建一个实例对象，反之 unapply 方法接受一个实例对象然后返回最初创建它所用的参数。提取器常用在模式匹配和偏函数中。

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

 def main(args: Array[String]): Unit = {
   // 调用工厂构造方法，构造出对象实例
   val person = Person("Spark", 6)
   // 这种写法居然可以正常编译
   val Person(name, age) = person
   /*
    * 使用了提取器，
    * 调用了unapply方法，把实例person中的name和age提取出来赋值给了Person类
    */
   println(name + " : " + age) //正常输出: Spark 6
   
   person match {
     // match过程就是调用提取器的过程
     case Person(name, age) => println("Wow, " + name + " : " + age)
   }
 }

自定义unapply方法
主要用于模式匹配中。

class Person(val name: String, val salary: Int)

object Person {

  def apply(name: String, salary: Int):Person = {
    new Person(name, salary)
  }

  def unapply(money: Person): Option[(String, Int)] = {
    if(money == null) {
      None
    } else {
      Some(money.name, money.salary)
    }
  }
  def main(args: Array[String]): Unit = {
    val person = Person("spark", 800);
     person match {
      // match过程就是调用提取器的过程
      case Person(name, age) => println("Wow, " + name + " : " + age)
    }
  }
}

for 表达式

Scala 提供一个轻量级的标记方式用来表示序列推导。推导使用形式为 for (enumerators) yield e 的 for 表达式，此处 enumerators 指一组以分号分隔的枚举器。一个 enumerator 要么是一个产生新变量的生成器，要么是一个过滤器。for 表达式在枚举器产生的每一次绑定中都会计算 e 值，并在循环结束后返回这些值组成的序列。
例子

case class User(name: String, age: Int)

val userBase = List(User("Travis", 28),
  User("Kelly", 33),
  User("Jennifer", 44),
  User("Dennis", 23))

val twentySomethings = for (user <- userBase if (user.age >=20 && user.age < 30))
  yield user.name  

twentySomethings.foreach(name => println(name))

这里 for 循环后面使用的 yield 语句实际上会创建一个 List。因为当我们说 yield user.name 的时候，它实际上是一个 List[String]。 user <- userbase="" 是生成器，if="" (user.age="">=20 && user.age < 30) 是过滤器用来过滤掉那些年龄不是20多岁的人。

泛型

泛型类指可以接受类型参数的类。泛型类在集合类中被广泛使用。
泛型类使用方括号 [] 来接受类型参数。一个惯例是使用字母 A 作为参数标识符，当然你可以使用任何参数名称。

定义一个泛型类

泛型类使用方括号 [] 来接受类型参数。一个惯例是使用字母 A 作为参数标识符，当然你可以使用任何参数名称。

class Stack[A] {
  private var elements: List[A] = Nil
  def push(x: A) { elements = x :: elements }
  def peek: A = elements.head
  def pop(): A = {
    val currentTop = peek
    elements = elements.tail
    currentTop
  }
}

上面的 Stack 类的实现中接受类型参数 A。这表示其内部的列表，var elements: List[A] = Nil，只能够存储类型 A 的元素。方法 def push 只接受类型 A 的实例对象作为参数(注意：elements = x :: elements 将 elements 放到了一个将元素 x 添加到 elements 的头部而生成的新列表中)。

使用

要使用一个泛型类，将一个具体类型放到方括号中来代替 A。

val stack = new Stack[Int]
stack.push(1)
stack.push(2)
println(stack.pop)  // prints 2
println(stack.pop)  // prints 1

型变

型变是复杂类型的子类型关系与其组件类型的子类型关系的相关性。 Scala支持泛型类的类型参数的型变注释，允许它们是协变的，逆变的，或在没有使用注释的情况下是不变的。在类型系统中使用型变允许我们在复杂类型之间建立直观的连接，而缺乏型变则会限制类抽象的重用性。

1
2
3

class Foo[+A] // 一个协变类
class Bar[-A] // 一个逆变类
class Baz[A]  // 一个不变类

协变

使用注释 +A，可以使一个泛型类的类型参数 A 成为协变。对于某些类 class List[+A]，使 A 成为协变意味着对于两种类型 A 和 B，如果 A 是 B 的子类型，那么 List[A] 就是 List[B] 的子类型。这允许我们使用泛型来创建非常有用和直观的子类型关系。

考虑以下简单的类结构：

abstract class Animal {
  def name: String
}
case class Cat(name: String) extends Animal
case class Dog(name: String) extends Animal

类型 Cat 和 Dog 都是 Animal 的子类型。 Scala 标准库有一个通用的不可变的类 sealed abstract class List[+A]，其中类型参数 A 是协变的。这意味着 List[Cat] 是 List[Animal]，List[Dog] 也是 List[Animal]。直观地说，猫的列表和狗的列表都是动物的列表是合理的，你应该能够用它们中的任何一个替换 List[Animal]。

在下例中，方法 printAnimalNames 将接受动物列表作为参数，并且逐行打印出它们的名称。如果 List[A] 不是协变的，最后两个方法调用将不能编译，这将严重限制 printAnimalNames 方法的适用性。

object CovarianceTest extends App {
  def printAnimalNames(animals: List[Animal]): Unit = {
    animals.foreach { animal =>
      println(animal.name)
    }
  }

  val cats: List[Cat] = List(Cat("Whiskers"), Cat("Tom"))
  val dogs: List[Dog] = List(Dog("Fido"), Dog("Rex"))

  printAnimalNames(cats)
  // Whiskers
  // Tom
  printAnimalNames(dogs)
  // Fido
  // Rex
}

逆变

通过使用注释 -A，可以使一个泛型类的类型参数 A 成为逆变。与协变类似，这会在类及其类型参数之间创建一个子类型关系，但其作用与协变完全相反。也就是说，对于某个类 class Writer[-A] ，使 A 逆变意味着对于两种类型 A 和 B，如果 A 是 B 的子类型，那么 Writer[B] 是 Writer[A] 的子类型。

考虑在下例中使用上面定义的类 Cat，Dog 和 Animal ：

1
2
3

abstract class Printer[-A] {
  def print(value: A): Unit
}

这里 Printer[A] 是一个简单的类，用来打印出某种类型的 A。让我们定义一些特定的子类：

class AnimalPrinter extends Printer[Animal] {
  def print(animal: Animal): Unit =
    println("The animal's name is: " + animal.name)
}

class CatPrinter extends Printer[Cat] {
  def print(cat: Cat): Unit =
    println("The cat's name is: " + cat.name)
}

如果 Printer[Cat] 知道如何在控制台打印出任意 Cat，并且 Printer[Animal] 知道如何在控制台打印出任意 Animal，那么 Printer[Animal] 也应该知道如何打印出 Cat 就是合理的。反向关系不适用，因为 Printer[Cat] 并不知道如何在控制台打印出任意 Animal。因此，如果我们愿意，我们应该能够用 Printer[Animal] 替换 Printer[Cat]，而使 Printer[A] 逆变允许我们做到这一点。

object ContravarianceTest extends App {
  val myCat: Cat = Cat("Boots")

  def printMyCat(printer: Printer[Cat]): Unit = {
    printer.print(myCat)
  }

  val catPrinter: Printer[Cat] = new CatPrinter
  val animalPrinter: Printer[Animal] = new AnimalPrinter

  printMyCat(catPrinter)
  printMyCat(animalPrinter)
}

这个程序的输出如下：

The cat’s name is: Boots
The animal’s name is: Boots

不变

默认情况下，Scala中的泛型类是不变的。这意味着它们既不是协变的也不是逆变的。在下例中，类 Container 是不变的。 Container[Cat] 不是 Container[Animal]，反之亦然。

class Container[A](value: A) {
  private var _value: A = value
  def getValue: A = _value
  def setValue(value: A): Unit = {
    _value = value
  }
}

可能看起来一个 Container[Cat] 自然也应该是一个 Container[Animal]，但允许一个可变的泛型类成为协变并不安全。在这个例子中，Container 是不变的非常重要。假设 Container 实际上是协变的，下面的情况可能会发生：

val catContainer: Container[Cat] = new Container(Cat("Felix"))
val animalContainer: Container[Animal] = catContainer
animalContainer.setValue(Dog("Spot"))
val cat: Cat = catContainer.getValue

糟糕，我们最终会将一只狗作为值分配给一只猫
幸运的是，编译器在此之前就会阻止我们。

上界

在Scala中，类型参数和抽象类型都可以有一个类型边界约束。这种类型边界在限制类型变量实际取值的同时还能展露类型成员的更多信息。比如像T <: A这样声明的类型上界表示类型变量T应该是类型A的子类。下面的例子展示了类PetContainer的一个类型参数的类型上界。
在Scala中，类型参数和抽象类型都可以有一个类型边界约束。这种类型边界在限制类型变量实际取值的同时还能展露类型成员的更多信息。比如像T <: A这样声明的类型上界表示类型变量T应该是类型A的子类。下面的例子展示了类PetContainer的一个类型参数的类型上界。

abstract class Animal {
 def name: String
}

abstract class Pet extends Animal {}

class Cat extends Pet {
  override def name: String = "Cat"
}

class Dog extends Pet {
  override def name: String = "Dog"
}

class Lion extends Animal {
  override def name: String = "Lion"
}

class PetContainer[P <: Pet](p: P) {
  def pet: P = p
}

val dogContainer = new PetContainer[Dog](new Dog)
val catContainer = new PetContainer[Cat](new Cat)
// this would not compile
val lionContainer = new PetContainer[Lion](new Lion)

类PetContainer接受一个必须是Pet子类的类型参数P。因为Dog和Cat都是Pet的子类，所以可以构造PetContainer[Dog]和PetContainer[Cat]。但在尝试构造PetContainer[Lion]的时候会得到下面的错误信息：

1	type arguments [Lion] do not conform to class PetContainer's type parameter bounds [P <: Pet]

这是因为Lion并不是Pet的子类。

下界

类型上界将类型限制为另一种类型的子类型，而类型下界将类型声明为另一种类型的超类型。术语 B >: A 表示类型参数 B 或抽象类型 B 是类型 A 的超类型。在大多数情况下，A 将是类的类型参数，而 B 将是方法的类型参数。

下面看一个适合用类型下界的例子：
下面看一个适合用类型下界的例子：

trait Node[+B] {
  def prepend(elem: B): Node[B]
}

case class ListNode[+B](h: B, t: Node[B]) extends Node[B] {
  def prepend(elem: B): ListNode[B] = ListNode(elem, this)
  def head: B = h
  def tail: Node[B] = t
}

case class Nil[+B]() extends Node[B] {
  def prepend(elem: B): ListNode[B] = ListNode(elem, this)
}

该程序实现了一个单链表。 Nil 表示空元素（即空列表）。 class ListNode 是一个节点，它包含一个类型为 B (head) 的元素和一个对列表其余部分的引用 (tail)。 class Node 及其子类型是协变的，因为我们定义了 +B。

但是，这个程序不能编译，因为方法 prepend 中的参数 elem 是协变的 B 类型。这会出错，因为函数的参数类型是逆变的，而返回类型是协变的。

要解决这个问题，我们需要将方法 prepend 的参数 elem 的型变翻转。我们通过引入一个新的类型参数 U 来实现这一点，该参数具有 B 作为类型下界。

trait Node[+B] {
  def prepend[U >: B](elem: U): Node[U]
}

case class ListNode[+B](h: B, t: Node[B]) extends Node[B] {
  def prepend[U >: B](elem: U): ListNode[U] = ListNode(elem, this)
  def head: B = h
  def tail: Node[B] = t
}

case class Nil[+B]() extends Node[B] {
  def prepend[U >: B](elem: U): ListNode[U] = ListNode(elem, this)
}

现在我们像下面这么做：

trait Bird
case class AfricanSwallow() extends Bird
case class EuropeanSwallow() extends Bird
val africanSwallowList= ListNode[AfricanSwallow](AfricanSwallow(), Nil())
val birdList: Node[Bird] = africanSwallowList
birdList.prepend(new EuropeanSwallow)

可以为 Node[Bird] 赋值 africanSwallowList，然后再加入一个 EuropeanSwallow。

抽象类型

特质和抽象类可以包含一个抽象类型成员，意味着实际类型可由具体实现来确定。例如：

trait Buffer {
  type T
  val element: T
}
}

这里定义的抽象类型T是用来描述成员element的类型的。通过抽象类来扩展这个特质后，就可以添加一个类型上边界来让抽象类型T变得更加具体。
特质和抽象类可以包含一个抽象类型成员，意味着实际类型可由具体实现来确定。例如：

abstract class SeqBuffer extends Buffer {
  type U
  type T <: Seq[U]
  def length = element.length
}

注意这里是如何借助另外一个抽象类型U来限定类型上边界的。通过声明类型T只可以是Seq[U]的子类（其中U是一个新的抽象类型），这个SeqBuffer类就限定了缓冲区中存储的元素类型只能是序列。

含有抽象类型成员的特质或类（classes）经常和匿名类的初始化一起使用。为了能够阐明问题，下面看一段程序，它处理一个涉及整型列表的序列缓冲区。

abstract class IntSeqBuffer extends SeqBuffer {
  type U = Int
}


def newIntSeqBuf(elem1: Int, elem2: Int): IntSeqBuffer =
  new IntSeqBuffer {
       type T = List[U]
       val element = List(elem1, elem2)
     }
val buf = newIntSeqBuf(7, 8)
println("length = " + buf.length)
println("content = " + buf.element)

这里的工厂方法newIntSeqBuf使用了IntSeqBuf的匿名类实现方式，其类型T被设置成了List[Int]。

把抽象类型成员转成类的类型参数或者反过来，也是可行的。如下面这个版本只用了类的类型参数来转换上面的代码：

abstract class Buffer[+T] {
  val element: T
}
abstract class SeqBuffer[U, +T <: Seq[U]] extends Buffer[T] {
  def length = element.length
}

def newIntSeqBuf(e1: Int, e2: Int): SeqBuffer[Int, Seq[Int]] =
  new SeqBuffer[Int, List[Int]] {
    val element = List(e1, e2)
  }

val buf = newIntSeqBuf(7, 8)
println("length = " + buf.length)
println("content = " + buf.element)

需要注意的是为了隐藏从方法newIntSeqBuf返回的对象的具体序列实现的类型，这里的型变标号（+T <: Seq[U]）是必不可少的。此外要说明的是，有些情况下用类型参数替换抽象类型是行不通的。

复合类型

需求:
有时需要表明一个对象的类型是其他几种类型的子类型。在 Scala 中，这可以表示成复合类型，即多个类型的交集。

案例：
假设我们有两个特质 Cloneable 和 Resetable：

trait Cloneable extends java.lang.Cloneable {
  override def clone(): Cloneable = {
    super.clone().asInstanceOf[Cloneable]
  }
}
trait Resetable {
  def reset: Unit
}

现在假设我们要编写一个方法 cloneAndReset，此方法接受一个对象，克隆它并重置原始对象：

def cloneAndReset(obj: ?): Cloneable = {
  val cloned = obj.clone()
  obj.reset
  cloned
}

这里出现一个问题，参数 obj 的类型是什么。如果类型是 Cloneable 那么参数对象可以被克隆 clone，但不能重置 reset; 如果类型是 Resetable 我们可以重置 reset 它，但却没有克隆 clone 操作。为了避免在这种情况下进行类型转换，我们可以将 obj 的类型同时指定为 Cloneable 和 Resetable。这种复合类型在 Scala 中写成：Cloneable with Resetable。

以下是更新后的方法：

1
2
3

def cloneAndReset(obj: Cloneable with Resetable): Cloneable = {
  //...
}

复合类型可以由多个对象类型构成，这些对象类型可以有单个细化，用于缩短已有对象成员的签名。格式为：A with B with C … { refinement }

关于使用细化的例子参考通过混入（mixin）来组合类。

自类型

自类型用于声明一个特质必须混入其他特质，尽管该特质没有直接扩展其他特质。这使得所依赖的成员可以在没有导入的情况下使用。

自类型是一种细化 this 或 this 别名之类型的方法。语法看起来像普通函数语法，但是意义完全不一样。

要在特质中使用自类型，写一个标识符，跟上要混入的另一个特质，以及 =>（例如 someIdentifier: SomeOtherTrait =>）。

trait User {
  def username: String
}

trait Tweeter {
  this: User =>  // 重新赋予 this 的类型
  def tweet(tweetText: String) = println(s"$username: $tweetText")
}

class VerifiedTweeter(val username_ : String) extends Tweeter with User {  // 我们混入特质 User 因为 Tweeter 需要
	def username = s"real $username_"
}

val realBeyoncé = new VerifiedTweeter("Beyoncé")
realBeyoncé.tweet("Just spilled my glass of lemonade")  // 打印出 "real Beyoncé: Just spilled my glass of lemonade"

因为我们在特质 trait Tweeter 中定义了 this: User =>，现在变量 username 可以在 tweet 方法内使用。这也意味着，由于 VerifiedTweeter 继承了 Tweeter，它还必须混入 User（使用 with User）。
自类型别名

object Demo {
  self =>
  def sum(num1: Int, num2: Int): Int = {
    num1 + num2
  }

  def main(args: Array[String]): Unit = {
    println(self.sum(1, 2))
  }
}

隐式转换

定义：指的是那种以implicit关键字声明的带有单个参数的函数。
通过隐式转换，程序员可以在编写Scala程序时故意漏掉一些信息，让编译器去尝试在编译期间自动推导出这些信息来，这种特性可以极大的减少代码量，忽略那些冗长，过于细节的代码。
1.将方法或变量标记为implicit
2.将方法的参数列表标记为implicit
3.将类标记为implicit

Scala支持两种形式的隐式转换：
隐式值：用于给方法提供参数
隐式视图：用于类型间转换或使针对某类型的方法能调用成功

案例

给File类增加read方法

1
2
3

class RichFile(val f: File) {
  def read() = Source.fromFile(f).mkString
}

门面类

import java.io.File

object RichFilePredef {
  implicit def fileToRichFile(f: File) = new RichFile(f)
}

使用

import java.io.File
import scala.io.Source
import com.tm.scala.implic.RichFilePredef._
object RichFile {
  def main(args: Array[String]) {
     val f = new File("/home/hadoop/sample_movielens_data.txt")
    // 注意：要使用隐式转换，必须在当前object之前导入隐式转换的门面object
    val contents = f.read()
    println(contents)
  }
}

排序中的使用案例

Ordered方式

1	class Girl(val name: String, var faceValue: Int, var age: Int)

定义比较规则

/**
  * 视图定界
  * 使用视图定界,视图定界其实就是隐式转换,将T转换成Ordered
  * 有时候，你并不需要指定一个类型是等/子/超于另一个类，你可以通过转换这个类来伪装这种关联关系。
  * 一个视界指定一个类型可以被“看作是”另一个类型。这对对象的只读操作是很有用的。
  * 更多知识：https://twitter.github.io/scala_school/zh_cn/advanced-types.html
  */
class OrderedChooser[T <% Ordered[T]] {
  def choose(first: T, second: T): T = {
    if (first > second) first else second
  }
}

门面类

object OrderedPredef {
  // 方式一
 implicit def gilrToOrdered(girl: Girl):Ordered[Girl] = new Ordered[Girl] {
        override def compare(that: Girl): Int = {
          if (girl.faceValue == that.faceValue) {
            girl.age - that.age
          } else {
            girl.faceValue - that.faceValue
          }
        }
      }

  // 方式二
  implicit val gilrToOrdered = (girl: Girl) => new Ordered[Girl] {
    override def compare(that: Girl): Int = {
      if (girl.faceValue == that.faceValue) {
        girl.age - that.age
      } else {
        girl.faceValue - that.faceValue
      }
    }
  }
}

测试类

object TestOrdered {
  def main(args: Array[String]): Unit = {
    val girl1 = new Girl("spark", 100,50)
    val girl2 = new Girl("mxnet", 90,30)
    import OrderedPredef._
    val chooser = new OrderingChoose[Girl]
    val g = chooser.choose(girl1, girl2)
    println(g.faceValue)
  }
}

Ordering方式

定义比较规则

/**
  * 文本定界
  */

class OrderingChoose[T: Ordering] {
  def choose(first: T, second: T): T = {
    val ord = implicitly[Ordering[T]]
    if (ord.gt(first, second)) first else second
  }
}

门面类

object OrderingPredef {
  //  方式一
  implicit val gilrToOrdering = new Ordering[Girl] {
    override def compare(x: Girl, y: Girl): Int = {
      x.faceValue - y.faceValue
    }
  }

  //  方式二
  implicit object GilrToOrdering extends Ordering[Girl] {
    override def compare(x: Girl, y: Girl): Int = {
      x.faceValue - y.faceValue
    }
  }

  // 方式三
  /**
    * 参考：Ordering中的下面方法
    * trait IntOrdering extends Ordering[Int] {
    * def compare(x: Int, y: Int) =
    * if (x < y) -1
    * else if (x == y) 0
    * else 1
    * }
    * implicit object Int extends IntOrdering
    */

  trait GirlToOrdering extends Ordering[Girl] {
    override def compare(x: Girl, y: Girl): Int = {
      x.faceValue - y.faceValue
    }
  }

  implicit object Girl extends GirlToOrdering

}

测试类

object TestOrdering {
  def main(args: Array[String]): Unit = {
    val g1 = new Girl("zhangsan", 50,50)
    val g2 = new Girl("lisi", 500,50)
    import OrderingPredef._
    val choose = new OrderingChoose[Girl]
    val g = choose.choose(g1, g2)
    println(g.name)
  }
}

scala导入类并取别名

导入Map,并取别名

1	import java.util.{Map => JMap}

scala Try的使用

spark Utils中的使用案例：

def classIsLoadable(clazz: String): Boolean = {
   // scalastyle:off classforname
   Try {
     Class.forName(clazz, false, getContextOrSparkClassLoader)
   }.isSuccess
   // scalastyle:on classforname
 }

quasiquotes(q字符串)用于代码生成

官方文档： https://docs.scala-lang.org/overviews/quasiquotes/intro.html
参考文档：https://www.cnblogs.com/shishanyuan/p/8455786.html
作用： Quasiquotes允许在Scala语言中对抽象语法树（AST）进行编程式构建，然后在运行时将其提供给Scala编译器以生成字节码。
以q开头的字符串是quasiquotes，虽然它们看起来像字符串，但它们在编译时由Scala编译器解析，并代表其代码的AST。

val tree = q"i am { a quasiquote }"  
tree: universe.Tree = i.am(a.quasiquote)

符号详解

== 和===的区别

参考：http://landcareweb.com/questions/25833/scala-sparkzhong-he-zhi-jian-de-qu-bie

== 返回一个布尔值
=== 返回一列（包含两列元素比较的结果）

call-by-value and call-by-name

传值调用（call-by-value）：先计算参数表达式的值，再应用到函数内部；
传名调用（call-by-name）：将未计算的参数表达式直接应用到函数内部

半生对象

object中的构造器在第一次调用执行一次，以后调用的话不会多次执行。
object会有自己的构造方法，默认是没有参数的构造方法。

scala自定注解

发表于 2019-01-04 | 分类于 scala |

| 字数统计: 279 | 阅读时长 ≈ 1

为什么要注解

除了编译和允许之外，还可以对程序做：
1、使用Scaladoc 自动产生文档;
2、漂亮的打印印出符合你偏爱分割的代码;
3、代码常见错误检查，如：打开了文件却没(在全部逻辑分支中)关闭。
4、实验类型检查，例如副作用管理或所有权属性确认。
这类工具（程序）被称为元编程工具，它们把其它程序当做输入程序。

scala注解所在包和基本语法格式

注解所在包：标准库定义的注解相关内容在包scala.annotation中。
基本语法： @注解名称(注解参数)

自定义注解

自定义注解需要从注解特质继承，scala提供两种注解：
1、基本语法： @注解名称(注解参数)
scala中的自定义注解不是接口/特质，而是类。
自定义注解需要从注解特质中继承，Scala中提供了两类注解特质：

scala.annotation.ClassfileAnnotation 由Java编译器生成注解
scala.annotation.StaticAnnotation 由Scala编译器生成注解

scala通过反射获取当前类类名

发表于 2019-01-04 | 分类于 scala |

| 字数统计: 979 | 阅读时长 ≈ 6

先编写HelloWord.scala文件

object HelloWorld {
  def main(args: Array[String]) {
    // 获取当前类类名
    println(this.getClass.getName.stripSuffix("$"))
  }
}

命令行使用scalac HelloWorld.scala编译后产生两个文件分别为HelloWorld.class和HelloWorld$.class

spark源码中的案例

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.spark.internal

import org.apache.log4j.{Level, LogManager, PropertyConfigurator}
import org.slf4j.{Logger, LoggerFactory}
import org.slf4j.impl.StaticLoggerBinder

import org.apache.spark.util.Utils

/**
 * Utility trait for classes that want to log data. Creates a SLF4J logger for the class and allows
 * logging messages at different levels using methods that only evaluate parameters lazily if the
 * log level is enabled.
 */
private[spark] trait Logging {

  // Make the log field transient so that objects with Logging can
  // be serialized and used on another machine
  @transient private var log_ : Logger = null

  // Method to get the logger name for this object
  protected def logName = {
    // Ignore trailing $'s in the class names for Scala objects
    this.getClass.getName.stripSuffix("$")
  }

  // Method to get or create the logger for this object
  protected def log: Logger = {
    if (log_ == null) {
      initializeLogIfNecessary(false)
      log_ = LoggerFactory.getLogger(logName)
    }
    log_
  }

  // Log methods that take only a String
  protected def logInfo(msg: => String) {
    if (log.isInfoEnabled) log.info(msg)
  }

  protected def logDebug(msg: => String) {
    if (log.isDebugEnabled) log.debug(msg)
  }

  protected def logTrace(msg: => String) {
    if (log.isTraceEnabled) log.trace(msg)
  }

  protected def logWarning(msg: => String) {
    if (log.isWarnEnabled) log.warn(msg)
  }

  protected def logError(msg: => String) {
    if (log.isErrorEnabled) log.error(msg)
  }

  // Log methods that take Throwables (Exceptions/Errors) too
  protected def logInfo(msg: => String, throwable: Throwable) {
    if (log.isInfoEnabled) log.info(msg, throwable)
  }

  protected def logDebug(msg: => String, throwable: Throwable) {
    if (log.isDebugEnabled) log.debug(msg, throwable)
  }

  protected def logTrace(msg: => String, throwable: Throwable) {
    if (log.isTraceEnabled) log.trace(msg, throwable)
  }

  protected def logWarning(msg: => String, throwable: Throwable) {
    if (log.isWarnEnabled) log.warn(msg, throwable)
  }

  protected def logError(msg: => String, throwable: Throwable) {
    if (log.isErrorEnabled) log.error(msg, throwable)
  }

  protected def isTraceEnabled(): Boolean = {
    log.isTraceEnabled
  }

  protected def initializeLogIfNecessary(isInterpreter: Boolean): Unit = {
    if (!Logging.initialized) {
      Logging.initLock.synchronized {
        if (!Logging.initialized) {
          initializeLogging(isInterpreter)
        }
      }
    }
  }

  private def initializeLogging(isInterpreter: Boolean): Unit = {
    // Don't use a logger in here, as this is itself occurring during initialization of a logger
    // If Log4j 1.2 is being used, but is not initialized, load a default properties file
    val binderClass = StaticLoggerBinder.getSingleton.getLoggerFactoryClassStr
    // This distinguishes the log4j 1.2 binding, currently
    // org.slf4j.impl.Log4jLoggerFactory, from the log4j 2.0 binding, currently
    // org.apache.logging.slf4j.Log4jLoggerFactory
    val usingLog4j12 = "org.slf4j.impl.Log4jLoggerFactory".equals(binderClass)
    if (usingLog4j12) {
      val log4j12Initialized = LogManager.getRootLogger.getAllAppenders.hasMoreElements
      // scalastyle:off println
      if (!log4j12Initialized) {
        val defaultLogProps = "org/apache/spark/log4j-defaults.properties"
        Option(Utils.getSparkClassLoader.getResource(defaultLogProps)) match {
          case Some(url) =>
            PropertyConfigurator.configure(url)
            System.err.println(s"Using Spark's default log4j profile: $defaultLogProps")
          case None =>
            System.err.println(s"Spark was unable to load $defaultLogProps")
        }
      }

      if (isInterpreter) {
        // Use the repl's main class to define the default log level when running the shell,
        // overriding the root logger's config if they're different.
        val rootLogger = LogManager.getRootLogger()
        val replLogger = LogManager.getLogger(logName)
        val replLevel = Option(replLogger.getLevel()).getOrElse(Level.WARN)
        if (replLevel != rootLogger.getEffectiveLevel()) {
          System.err.printf("Setting default log level to \"%s\".\n", replLevel)
          System.err.println("To adjust logging level use sc.setLogLevel(newLevel). " +
            "For SparkR, use setLogLevel(newLevel).")
          rootLogger.setLevel(replLevel)
        }
      }
      // scalastyle:on println
    }
    Logging.initialized = true

    // Force a call into slf4j to initialize it. Avoids this happening from multiple threads
    // and triggering this: http://mailman.qos.ch/pipermail/slf4j-dev/2010-April/002956.html
    log
  }
}

private object Logging {
  @volatile private var initialized = false
  val initLock = new Object()
  try {
    // We use reflection here to handle the case where users remove the
    // slf4j-to-jul bridge order to route their logs to JUL.
    val bridgeClass = Utils.classForName("org.slf4j.bridge.SLF4JBridgeHandler")
    bridgeClass.getMethod("removeHandlersForRootLogger").invoke(null)
    val installed = bridgeClass.getMethod("isInstalled").invoke(null).asInstanceOf[Boolean]
    if (!installed) {
      bridgeClass.getMethod("install").invoke(null)
    }
  } catch {
    case e: ClassNotFoundException => // can't log anything yet so just fail silently
  }
}

tang

火星度假村追梦程序员。

RSS