keep running

基础知识

相信python2的编码问题大多数开发同学都遇到过，在出现非 ascii 编码字符时，就很容易编码异常的问题。python2的字符编码分为 str 以及 unicode，具体情况这里不再敖述，只会总结字符串拼接时应该注意的问题以及可能遇到的坑点。

以下几点常识是下面进一步讨论问题的基础：

• str转为unicode的过程，称为解码，即 decode。
• unicode转为str，称为编码，即 encode。
• 使用%把str和unicode拼接，会自动隐式地把str转为unicode后，再进行拼接。（如果是fomat拼接呢？这里留个悬念，答案稍后揭晓）
• 当导入__future__包的unicode_literals特性时，python定义的字符都是unicode，而不是默认的str。这个也是为了让python2能够导入python3的特性，因为在python3中的str都是unicode。

% 拼接字符串

我们首先看看pyhon2中的使用%字符串拼接情况。从第一组结果来看，我们可以看到只要格式化串和字符串参数其中一个为unicode，最终结果就为unicode，这个和上面讲的第三点一致。在对str和unicode拼接的时候，会自动把str转为unicode，如第二组的中间两个结果。

但是我们需要注意编码的问题，如第2个结果，由于”中文”是非 acsii 编码，而且python解释器不知道其类型，会用ascii编码对其进行解码，相当于 u"%s" % ("中文").decode("ascii")，而ascii不认识非 0~127 编码所以就报错，当然我们可以手动指定用”utf-8”进行解码。

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

>>> type("%s" % ("hello"))
<type 'str'>
>>> type(u"%s" % ("hello"))
<type 'unicode'>
>>> type("%s" % (u"hello"))
<type 'unicode'>
>>> type(u"%s" % (u"hello"))
<type 'unicode'>
>>>
>>> type("%s" % ("中文"))
<type 'str'>
>>> type(u"%s" % ("中文"))         # 最终结果为unicode，会隐式地通过ascii编码把"中文"解码为unicode
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe4 in position 0: ordinal not in range(128)
>>> type(u"%s" % ("中文".decode("utf-8")))
<type 'unicode'>
>>> type("%s" % (u"中文"))
<type 'unicode'>
>>> type(u"%s" % (u"中文"))
<type 'unicode'>

format 拼接字符串

同样的，我们先看下面的第一组结果，是不是有点吃惊？第一组的第二结果不是unicode类型，而是str类型，这个跟%是不同的。很显然从结果上看，我们知道对于format其拼接结果类型取决于其格式化串的类型，而与参数没有任何关系。

理解的第一组数据的规律后，再看第二组就知道为什么有的情况会报异常了。第二组的第二个结果，由于最终结果为str，pyhton解释器会默认用 ascii 对 u"中文" 进行编码，而第三个结果，由于最终结果为unicode，python解释器会默认用 ascii 对应 “中文” 进行解码，而报错的理由和前面的情况一致，都是因为ascii不认识非 0~127 编码。

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

>>> type("{}".format("hello"))
<type 'str'>
>>> type("{}".format(u"hello"))
<type 'str'>
>>> type(u"{}".format("hello"))
<type 'unicode'>
>>> type(u"{}".format(u"hello"))
<type 'unicode'>
>>>
>>> type("{}".format("中文"))
<type 'str'>
>>> type("{}".format(u"中文"))         # 最终结果为str，会隐式地通过ascii编码把u"中文"编码为ascii
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
UnicodeEncodeError: 'ascii' codec can't encode characters in position 0-1: ordinal not in range(128)
>>> type("{}".format(u"中文".encode("utf-8")))
<type 'str'>
>>> type(u"{}".format("中文"))          # 最终结果为unicode，会隐式地通过ascii编码把"中文"解码为unicdoe
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe4 in position 0: ordinal not in range(128)
>>> type(u"{}".format("中文".decode("utf-8")))
<type 'unicode'>
>>> type(u"{}".format(u"中文"))
<type 'unicode'>

坑点

而我们线上的问题，要比以上两种都要隐蔽，大致如下：

代码目录结构

1
2
3
4
5
6
7

 tree -L 2 -I "*.pyc"
.
├── test_module_002
│   ├── __init__.py
│   ├── __pycache__
│   ├── main.py
│   └── module_a.py

如无特别说明，本文讨论的内容均基于 flink 1.7.1

最近一段时间用 flink 写一些 etl 作业，做数据的收集清洗入库，也遇到一些性能问题需要进一步解决，于是计划学习部分flink底层知识。第一篇，跟以前学习spark一样，从flink的并行度说起。

flink作业的启动模式

通过 flink YARN Setup 文档我们能够了解到，flink的启动方式大致有两种，
一种是先分配jobmanager、taskmanager的资源，等待后续提交作业，另一种是在提交的时候申请资源并运行。下面将简单介绍一下这两种启动方式的区别，并着重关注其并行度的计算，最后和spark并行度的计算对对比。

部署方式一：在yarn中启动一个flink session，提交job到该session

• 启动flink session
  • ./bin/yarn-session.sh -tm 8192 -s 32
  • 关键配置：
    • -n，指定 container 数量（即taskmanager的数量，不过已经不建议使用，对应的源码
    • -tm，分配 taskmanager 内存大小
    • -jm，分配 jobmanager 内存大小
    • -s，每个taskmanager分配slot个数（如果配置了将会覆盖yarn的 parallelism.default 配置，parallelism.default 值默认为1）
    • -Dyarn.containers.vcores，在yarn中分配的vcore个数，默认和slot个数一致，即一个slot一个vcore
    • 默认 taskmanager 的数量为1，并行度为 slot * taskmanager ，源码
  • 一旦 flink session在yarn中启动成功，将会展示有关 jobmanager 连接的详细信息，通过CTRL+C 或者 在client中输入stop关闭 flink session
• 提交job到该session
  • ./bin/flink run ./examples/batch/WordCount.jar
  • 关键配置：
    • -c，指定入口class
    • -m，指定jobmanager地址
    • -p，指定作业的并行度
  • client能够自动识别对应的 jobmanager 地址
  • 并行度的确定：
    • 如果不指定 -p ，则作业并行度为 1 （parallelism.default 的配置值，默认为1）
    • 如果指定-p，则作业则在该session下，以 -p 指定值的并行度运行。如果作业的并行度大于session的并行度，则会报异常，作业启动失败。

部署方式二：在yarn中启动一个单独的作业

• ./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar
• flink session的配置同样适用于启动单独的作业，需要加前缀 y 或者 yarn
• 关键配置：
  • -n ，允许加载savepoint失败时启动程序
  • -d，client非阻塞模式启动作业
  • -p，指定作业并行度
  • -ytm，分配 taskmanager 内存大小
  • -yjm，分配 jobmanager 内存大小
  • -ys，指定每个taskmanager分配slot个数
  • -yn，指定container数量，和taskmanager数量一致
• 并行度的确定
  • 如果指定了-m yarn-cluster，并且是 -d 或者 -yd 模式，不通过 -yid 指定 applicationid，则其并行度由 -p 决定。
  • flink会启动多少个taskmanager？我们知道flink作业的实际并行度是由 taskmanager * slot 决定的，默认情况下每个taskmanager的slot数量为1，所以yarn最终为了实现并行度为 -p 的作业，需要启动p个taskmanager。num( taskmanenger ) = p / slot

spark on yarn vs. flink on yarn

spark相关的executor以及并行的计算见 Spark学习系列之一和之二

• executor vs. taskmanager
  • spark submit 通过 –num-executors 控制executor数量
  • flink run 通过 -p 和 -ys 控制taskmanager数量

另外spark on standalone模式下，其executor数量的计算方式和flink run差不多，它也是通过总的核数和每个executor核数反算所需的executor数目，可以把 total-executor-cores 类比 -p，executor-cores 类比 -ys）

如无特别说明，本文讨论的内容均基于 es 7.*

es的Rollover索引

es的Rollover索引通常指的是一个别名指向某个索引，并且能够在索引的某些条件下进行轮转，如索引的创建时间长短、大小、文档数量。

如创建一个名为 nginx_log-000001 的索引，并指定其alias为nginx_log_write，并且我们对nginx_log_write写入3个文档（其实也是对nginx_log-000001写）。然后对别名调用rollover接口，
由于已经达到文档数目为3的条件，则会自动生成 nginx_log-000002 的索引。这时对nginx_log_write写入会自动写入到nginx_log-000002索引中。

需要注意的是，由于对索引设置alias的时候，没有添加 "is_write_index": true 配置，则在执行rollover并创建新索引成功后，将会只指向一个索引（新索引），对nginx_log_write查询只能查到最新索引的数据，而不能查到历史数据。相反，如果配置了"is_write_index": true，在rollover后alias会同时指向多个索引，并且最新索引设置为"is_write_index": true，旧索引设置为"is_write_index": false，对alias的
写入就是对最新索引的写入，查询时是对所有索引进行查询。

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
46
47
48

# 创建索引nginx_log-000001，并设置其别名为nginx_log_write
PUT /nginx_log-000001
{
  "aliases": {
    "nginx_log_write": {
    }
  }
}

# 对别名写入文档，重复执行3次
POST nginx_log_write/_doc
{
  "log":"something before rollover"
}

# 对别名执行rollover
POST /nginx_log_write/_rollover
{
  "conditions": {
    "max_age":   "1d",
    "max_docs":  3,
    "max_size":  "5gb"
  }
}

# 对新索引插入新数据
POST nginx_log_write/_doc
{
  "log":"something after rollover"
}

# 分别查 nginx_log-000001、nginx_log-000002、nginx_log_write，对ginx_log_write只能查到最新索引的数据
POST nginx_log_write/_search
{
  "query":{
    "match_all": {
    }
  }
}

# 对非 "is_write_index": true 模式的索引，可用 index_name-* 查询所有数据
POST nginx_log-*/_search
{
  "query":{
    "match_all": {
    }
  }
}

另外，我们可以利用 Date Math 创建带日期的rollover索引，更加方便索引管理。

1
2
3
4
5
6
7
8
9

# PUT /<nginx_log-{now/d}-000001>，将创建名为 nginx_log-2020.05.17-000001 的索引
PUT /%3Cnginx_log-%7Bnow%2Fd%7D-000001%3E
{
  "aliases": {
    "nginx_log_write": {
    }
  }
}

需要注意的是 _rollover api只会对调用该接口的那个时刻有效，当然可以自己独立做一个任务周期性扫描所有别名，当别名到达一定条件后就调用其 _rollover 接口。如果需要es自身定时调用的话，可以使用自动化程度更高的 Index Lifecycle Management。

Index Lifecycle Management

与 _rollover 索引相比，索引生命周期管理会更加自动化，ILM把索引的生命周期分为4个phase，分别为Hot、Warm、Cold、Delete。每个phase可以包含多个action。

如无特别说明，本文源码版本为 spark 2.3.4
两个rdd join时产生新的rdd，是宽依赖，还是窄依赖？

join transformation

以上图片是个经常用来解释宽窄依赖的经典图，来源于论文<<Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing>>。以下这段话也来自与该论文：

join: Joining two RDDs may lead to either two nar- row dependencies (if they are both hash/range partitioned with the same partitioner), two wide dependencies, or a mix (if one parent has a partitioner and one does not). In either case, the output RDD has a partitioner (either one inherited from the parents or a default hash partitioner)

或许我们会好奇，为什么同样是join操作，有时是宽依赖，有时窄依赖？我们先从两个简单的实验开始，再从源码看其实现方式。

rdd1和rdd2的partitioner不同

假设我们有rdd1和rdd2，其partitioner分别为partitioner1、partitioner2。分区器定义如下：

partitioner1：

1
2

numPartiton = 3
func = x mod numPartiton

partitioner2：

1
2

numPartiton = 5
func = (x  * 3) mod numPartiton

rdd1的初始分布如下：

1
2
3

partition0: (0, "a"), (3, "e")
partition1: (1, "b")
partition2: (2, "c")

rdd2的初始分布如下：

1
2
3
4
5

partition0: (0, "e"), (0, "j")
partition1: (2, "f")
partition2: (4, "g")
partition3: (1, "h"), (6, "k")
partition4: (3, "i")

rdd3=rdd2.join(rdd1)，rdd3数据分布如下：

1
2
3
4
5

partition0: (0, ("e", "a")), (0, ("j", "a"))
partition1: (2, ("f", "c"))
partition2: 
partition3: (1, ("h", "b"))
partition4: (3, ("i", "e"))

rdd3和rdd1以及rdd2的parittion之间的依赖关系如下：

1
2
3
4
5
6
7
8
9

rdd1.partition0 ==> rdd3.partition0, rdd3.partition4 
rdd1.partition1 ==> rdd3.partition3  
rdd1.partition2 ==> rdd3.partition2  

rdd2.partition0 ==> rdd3.partition0  
rdd2.partition1 ==> rdd3.partition1 
rdd2.partition2 ==> rdd3.partition2  
rdd2.partition3 ==> rdd3.partition3  
rdd2.partition4 ==> rdd3.partition4  

可以看到rdd1的parittion0 同时被rdd3的partition0和partition4依赖，父rdd的一个parittion被子rdd多个parittion依赖，所以此时rdd3对rdd1的依赖为宽依赖，而对rdd2为窄依赖。

版本情况dbutils:1.1; pymysql:0.9.3; python:2.7.13

线上情景

最近线上维护时，由于只需要更改数据库配置，所以就重启了数据库，而python应用没有重启。在重启数据库后，日志显示正常，也能成功入库。后来接到反馈表示有部分数据没有入库，紧急重启python应用，后续数据入库正常。而我则负责找出原因以及修复bug的工作。

调研原因

在排查完其他问题后，最异常的是对于有部分请求，日志显示处理成功了，但是却没入库，排查了好几天找不到原因。为此写了demo来帮助排查，为了可以自动commit，采用的是setsession=[“set autocommit=1”]方式设置每个底层的连接为自动提交。在测试demo期间，数据库重启后之后的sql就无法入库。demo代码如下：

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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# -*- coding: utf-8 -*-

import pymysql
import time
import traceback
from DBUtils.PooledDB import PooledDB
from pymysql import MySQLError

pymysql.install_as_MySQLdb()
con = None
pooledDB = None

def try_insert():
    i = 0
    while i < 60:
        print "============ {0} ============".format(i)
        # 除了第一次从库中拿，不用ping，直接接初始化链接
        # 后面如果有cache connection，则从cache中并且进行ping，如果失败则用_create()重新初始化connection
        # con 类型为 PooledDedicatedDBConnection
        # con._con 类型为 SteadyDBConnection
        # con._con._con 类型为 pymysql中的Connection类型
        # con._con._con._sock 类型为 mysql 连接
        con = pooledDB.connection()
        # con._con._con.autocommit(True)
        print "con._con id = {0}".format(id(con._con))
        print "con._con._con id = {0}".format(id(con._con._con))
        print "con._con._con._sock id = {0}".format(id(con._con._con._sock))
        try:
            cursor = con.cursor(pymysql.cursors.DictCursor)
            if not cursor:
                print "cursor is {0}".format(cursor)
            select_sql = "insert into user2(name,age) values('zhang', 20)"
            ret_rows = cursor.execute(select_sql)
            print cursor._last_executed
            print "ret_rows is {0}".format(ret_rows)

        except MySQLError as e:
            print "MySQLError error: {0}".format(e)
            print traceback.format_exc()
        except Exception as e:
            print "Exception error: {0}".format(e)
            print traceback.format_exc()

        i = i + 1
        time.sleep(1)
        con.close()


if __name__ == "__main__":
    db_conf = {'user':'root','passwd':'zhang','host':'127.0.0.1','port':3306,'connect_timeout':5,'db':'test_dbutils'}
    # db_conf = {'user':'root','passwd':'zhang','host':'127.0.0.1','port':3306,'connect_timeout':5,'db':'test_dbutils',"autocommit":True}


    pooledDB = PooledDB(
        creator=pymysql,  # 使用数据库连接的模块
        maxconnections=4,  # 连接池允许的最大连接数，0和None表示不限制连接数
        mincached=0,  # 初始化时，连接池中至少创建的空闲的链接，0表示不创建
        maxcached=0,  # 连接池中最多闲置的链接，0和None不限制
        maxshared=0,  # 连接池中最多共享的链接数量，0表示不共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，此值只有在creator.threadsafety > 1时设置才有效，否则创建的都是dedicated connection，即此连接是线程专用的。
        blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
        maxusage=None,  # 一个连接最多被重复使用的次数，None表示无限制
        setsession=["set autocommit=1"],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]；务必要设置autocommit，否则可能导致该session的sql未提交
        ping=1,  # 每次从pool中取连接时ping一次检查可用性
        reset=False,  # 每次将连接放回pool时，将未提交的内容回滚；False时只对事务操作进行回滚
        **db_conf
    )

    try_insert()

如无特别说明，本文源码版本为 spark 2.3.4

创建rdd有三种方式，一种是通过SparkContext.textFile()访问外部存储创建，一种是通过输入数据集合通过调用 SparkContext.parallelize() 方法来创建，最后一种是通过转换已有的rdd生成新的rdd。

通过parallelize创建rdd的分区数量分析

通过parallelize的方式比较简单，相信也是大部分初学者第一次接触创建rdd的方法，那么通过这个方法创建的rdd的默认分区数是多少呢？我们通过源码进行分析。

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

package org.apache.spark.SparkContext

class SparkContext(config: SparkConf) extends Logging {
  /** Default level of parallelism to use when not given by user (e.g. parallelize and makeRDD). */
  def defaultParallelism: Int = {
    assertNotStopped()
    taskScheduler.defaultParallelism
  }
  
  def parallelize[T: ClassTag](
      seq: Seq[T],
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    assertNotStopped()
    new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())
  }
}

我们先看看parallelize是如何生成rdd的。可以看到它是通过 ParallelCollectionRDD 类创建一个rdd，其内部返回的partitioner是通过ParallelCollectionRDD伴生对象的slice方法分割seq为一个二维的Seq[Seq[T]]，并把这个二维的序列传递到ParallelCollectionPartition中实例化的。

接下来是关键，defaultParallelism的默认值确定了分区的数量。

如无特别说明，本文源码版本为 spark 2.3.4

学习spark有一段时间了，最近想动动手写个demo出来，大致的功能是从kafka读取用户点击记录，用spark streaming对这些数据进行读取并统计用户一段时间的点击记录，期望最后能落盘到redis中供需求方调用。

这个demo看似简单，但是作为一个新手，我也遇到了一些看起来比较奇怪的问题。再此总结一下我遇到的一些问题，希望能给遇到同样问题的人带来一些帮助。

问题一：spark的并行度是多少？

我相信一开始接触的初学者对此肯定有疑惑，并行度指的什么？我认为在spark中，这个并行度指的是partition的数量，无论是通过parallelize初始化rdd，还是通过join和reduceByKey等shuffle操作，都意味着需要确定这个新rdd的paritition数量。这里涉及到一个参数spark.default.parallelism，该参数__大多数情况下__是parallelize、join、reducdeByKey等操作的__默认__并行度。如果不定义这个参数，默认情况下分区数量在不同情景的情况下有所不同：

• 对于join和reduceByKey等shuffle操作，分区数一般为多个父rdd中partition数目最大的一个。
• 对于parallelize进行初始化操作，分区数在不同部署模式下不同：
  • local[*]：本地cpu的core数量，local[N]则为N，local则为1
  • meos：默认为8
  • other：一般为executor个数 * 每个executor的core个数
• 当然如果定义了spark.default.parallelism参数，其默认分区数也不一定是其值，具体分析见Spark学习系列之二：rdd分区数量分析。实际api中也能通过传递numPartitions参数覆盖spark.default.parallelism，自行决定并行度。
• 比如正在使用的mac是四核，假设向yarn申请executor个数为2，每个executor的core数量为1，那么spark.default.parallelism的值为2，这时一般情况下是不能充分利用其申请核数资源的，最好是申请核数的23倍。可以通过 –conf 传入参数 --conf spark.default.parallelism = 4 或者 --conf spark.default.parallelism = 6，使其默认值为申请核数的23倍。如果有的task执行比较快，core就空闲出来了，为了多利用core就设置task数量为2~3倍。当然最后的并行度还需要根据实际情况进行分析。

如何确定本机核数？通过local[*]模式进行parallelize初始化rdd，再输出myrdd.partitions.size即可得，也可以通过java代码Runtime.getRuntime.availableProcessors()获得

参考：
https://spark.apache.org/docs/latest/configuration.html
http://spark.apache.org/docs/latest/tuning.html

问题二：standalone模式下，executor个数和executor核数如何确定？

由于需要通过spark streaming读取kafka，如果对应topic的partition数量已知，那么应该启动对应个数的executor，因为kafka的一个parition同一时间只允许同一个groupid的consumer读取，如果topic的partition为1，申请的executor为2，那么将只有一个executor的资源得到了利用。

既然executor个数比较重要，yarn模式可以通过--num-executors确定executor个数，那standalone模式如何确定的呢？直接先说结论：

• executor的数量 = total-executor-cores/executor-cores
• --total-executor-cores 对应配置 spark.cores.max (default: spark.deploy.defaultCores)，表示一个application最大能用的core数目；如果没有设置则默认上限为spark.deploy.defaultCores，该配置的值默认为infinite（不限）
• --executor-cores 对应配置 spark.executor.cores，表示每个executor的core数目
• 可以看到standalone的executor数量并不能直接指定，而是通过core的换算得到的，如果对executor数目有要求的话，可以额外关注一下。

以下是我写demo过程遇到问题，以及解决问题的大致流程。

在写demo过程中通过spark-sumbit提交任务时，忘了写master，但是通过--executor-cores指定了每个executor的core数量。等应用跑起来后，发现spark ui上，发现worker上有1个executor，每个executor4个core，这显然不符合的预期。明明通过--executor-cores指定了executor的core数量，为什么申请到的core数目不符合预期？即使spark-submit的script中没包含master，但是程序是指定了master（spark://zhangqilongdeMacBook-Air.local:7077）。我决定进行多次调整参数，验证每种情况下申请到executor数量和每个executor的core数量，总结如下：

• master和executor-cores，只配置一个或者两个都不配，则只申请一个executor，并且executor将尽量使用worker的所有core。
• master和executor-cores两个都配，则申请的executor数量 = workder core的总数/executor-cores，每个executor的core数量和executor-cores一致。

通过源码可以发现：

• --executor-cores只有在–master为standalone、yarn、kubernetes模式下才会生效，如果不是这些模式，将会通过__默认配置文件__指定缺失的值。即如果不指定master的情况下（默认为local[*]），--executor-cores并不会生效，并且使用 SPAKR_HOME/conf/spark-defaults.conf配置文件中的值对其赋值，如果该配置文件中依然不存在，则为spark系统默认对该变量的值，即infinite（不限）。
• --total-executor-cores可以配置standalone每个application可以用的核总数（其实通过spark-submit命令行的提示也能看出来，因为yarn模式下该值不可配所以一开始这个配置被我忽略了）

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

org.apache.spark.deploy.Submit

...省略部分代码
	//可以看到spark.executor.cores只在某些情况下才会被赋值
	OptionAssigner(args.executorCores, STANDALONE | YARN | KUBERNETES, ALL_DEPLOY_MODES, confKey = "spark.executor.cores"),
	OptionAssigner(args.totalExecutorCores, STANDALONE | MESOS | KUBERNETES, ALL_DEPLOY_MODES, confKey = "spark.cores.max")

...省略部分代码
    // Load any properties specified through --conf and the default properties file
    // 通过sparkProperties（已经读取了spark-defaluts.conf内动）hashMap对缺失配置进行填充。
    for ((k, v) <- args.sparkProperties) {
      sparkConf.setIfMissing(k, v)
    }

...省略部分代码

参考：
https://spark.apache.org/docs/latest/spark-standalone.html

在golang开发中，在使用mysql数据库时一般使用数据库驱动包为 go-sql-driver/mysql，该包是按照go官方包database/sql定义规范实现的。我们线上的程序偶尔会在标准错误输出 “broken pip”，为了究其原因做了些调研，并给出了解决方法。

线上场景

目前 项目A 使用的go-sql-driver/mysql版本为 3654d25ec346ee8ce71a68431025458d52a38ac0 ， 项目B 使用的版本为 v1.3.0 ，其中 项目A 的版本低于v1.3.0。它们线上标准错误输出都有类似以下的日志，但是程序的业务逻辑却没有影响。

1
2

[mysql] 2019/08/01 17:12:18 packets.go:33: unexpected EOF
[mysql] 2019/08/01 17:12:18 packets.go:130: write tcp 127.0.0.1:59722->127.0.0.1:3306: write: broken pipe

通过日志输出以及堆栈可以找到 go-sql-driver/mysql/packets.go 对应的源码，可以发现第一条日志是以下第8行代码打印，第二条是第9行调用mc.Close()关闭连接时报错。

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

// Read packet to buffer 'data'
func (mc *mysqlConn) readPacket() ([]byte, error) {
    var prevData []byte
    for {
        // read packet header
        data, err := mc.buf.readNext(4)
        if err != nil {
            errLog.Print(err)
            mc.Close()
            return nil, driver.ErrBadConn
        }
     // 省略部分代码
}

问题复现

通过网上搜索能够大概猜出是mysql server主动关闭的原因，我们可以通过设置mysql server主动关闭连接来复现线上场景，并且通过tcpdump观察其原因。

1.设置mysql server主动关闭连接时间

mysql server默认设置的关闭不活跃连接时间为28800秒（8小时），我们通过 set global wait_time=10 设置为10秒，便于问题重现。

2.运行tcpdump和测试demo

1.通过tcpdump可以收集tcp数据包的发送接收情况，尤其是的在mysql server关闭连接后，go程序如何和mysql server交互是我们关注的重点，tcpdump命令如下：

sudo tcpdump -s 0 -t -i lo -l port 3306 -w lo.cap

2.运行一个简单的测试demo

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

package main
import (
    "database/sql"
    "log"
    "time"
    _ "github.com/go-sql-driver/mysql"
)
func main() {
    // before you run this test program, please run the script in your mysql
    // "set global wait_timeout=10;"
    // 表示mysql server关闭不活跃连接的等待时间
    // 参考 https://github.com/go-sql-driver/mysql/issues/657
    db, err := sql.Open("mysql", "root:zhang@tcp(127.0.0.1:3306)/?charset=latin1&autocommit=1&parseTime=true&loc=Local&timeout=3s")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    //db.SetConnMaxLifetime(5 * time.Second)
    err = db.Ping()
    if err != nil {
        log.Fatal(err)
    }
    go func() {
        for {
            _, err := db.Exec("select * from test_time.A")
            if err != nil {
                log.Fatal(err)
            }
            // Wait for 11 seconds. This should be enough to timeout the conn, since `wait_timeout` is 10s
            time.Sleep(11 * time.Second)
        }
    }()
    time.Sleep(1000 * time.Second)
}

部门某些业务需要在海外上线，涉及到数据库时区、应用时区的转换。本文将讨论golang针对数据库时区的处理问题。

为了方便讨论，避免混淆，本文对“时间”的表达方式作出约定：时间=时区时间+时区。如时间 2019-05-21 15:48:38 CST ,则其时区时间为2019-05-21 15:48:38，时区为CST。如果没有特别说明，本文提到的“时间”都包含时区。

一、golang中mysql数据库驱动的时区配置

mysql中关于时间日期的概念数据模型有DATE、DATETIME、TIMESTAMP，golang程序根据数据链接DSN（Data Source Name）配置，数据库驱动 github.com/go-sql-driver/mysql 可以对这三种类型的值转换成go中的time.Time类型，关键配置如下：

• parseTime
  • 默认为false，把mysql中的 DATE、DATETIME、TIMESTAMP 转为golang中的[]byte类型
  • 设置为true，将会转为golang中的 time.Time 类型
• loc
  • 默认为UTC，表示转换DATE、DATETIME、TIMESTAMP 为 time.Time 时所使用的时区
  • 设置成Local，则与系统设置的时区一致
  • 如果想要设置成中国时区可以设置成 Asia/Shanghai ，更多的时区可以参考 /usr/share/zoneinfo/ 或者$GOROOT/lib/time/zoneinfo.zip。

在实际的使用中，我们往往会配置成 parseTime=true 和 loc=Local，这样避免了手动转换DATE、DATETIME、TIMESTAMP。

二、golang如何转换mysql的时间类型

在涉及到不同时区时，我们golang程序应该怎么处理mysql的 DATE、DATETIME、TIMESTAMP 数据类型？是否只要配置了parseTime=true&loc=xxx就不会有问题？我们来做两个小实验。

实验一：应用和数据库在同一时区

1.timestamp

a.系统时区设置为CST，mysql和golang在同一个时区的机器上。（如何设置和查看时区可以参考本文第五节内容。）

• golang在程序中连接数据库使用的配置DSN是parseTime=true&loc=xxx，xxx分别为UTC、Asia/Shanghai、Europe/London、Local。
• mysql终端中insert一条timestamp【时区时间】为2019-04-02 13:18:17的记录，其UNIX_TIMESTAMP(timestamp)=1554182297。

以下1~5行均为golang程序读取刚插入数据库的数据结果，第一列输出分别为链接数据库DSN配置，第二列为转换为time.Time后的输出。

1
2
3
4

parseTime=true&loc=UTC:                  2019-04-02 13:18:17 +0000 UTC
parseTime=true&loc=Asia/Shanghai:        2019-04-02 13:18:17 +0800 CST
parseTime=true&loc=Europe/London:        2019-04-02 13:18:17 +0100 BST
parseTime=true&loc=Local:                2019-04-02 13:18:17 +0800 CST

b.同样的机器，修改系统时区为BST，在mysql终端中select上一步插入的数据，timestamp【时区时间】为2019-04-02 06:18:17，UNIX_TIMESTAMP(timestamp)=1554182297。程序输出为：

1
2
3
4

parseTime=true&loc=UTC:                  2019-04-02 06:18:17 +0000 UTC
parseTime=true&loc=Asia/Shanghai:        2019-04-02 06:18:17 +0800 CST
parseTime=true&loc=Europe/London:        2019-04-02 06:18:17 +0100 BST
parseTime=true&loc=Local:                2019-04-02 06:18:17 +0100 BST

c.小结：

• UNIX_TIMESTAMP可以把mysql的timstamp转为距离 1970-01-01 00:00:00 UTC 的秒数，这个经过转换后的值无论mysql在任何时区都不会变。
• 即使同一条数据库记录，由于时区不同，mysql终端中直接select出的timestamp的【时区时间】也不同。也侧面说明了mysql内部实现的timstamp结构体中包含了时区信息，在输出时根据当前时区做转换，输出当前【时区时间】。
• golang程序获取到的time.Time等于：mysql【时区时间】+ 时区，时区为loc指定的时区，与mysql时区没有关系。

从Two-dimensional slices开始，使用中文版的effctive_go学习
https://www.kancloud.cn/kancloud/effective/72207

Data：

二维切片:

• Go的数组和切片都是一维的。要创建等价的二维数组或者切片，需要定义一个数组的数组或者切片的切片。

Maps:

• Map是一种方便，强大的内建数据结构，其将一个类型的值（key）与另一个类型的值（element或value） 关联一起。
• key可以为任何 定义了等于操作符 的类型，例如整数，浮点和复数，字符串，指针，接口（只要其动态类型支持等于操作），结构体和数组。
• 切片不能 作为map的key，因为它们没有定义等于操作。和切片类似，map持有对底层数据结构的引用。如果将map传递给函数，其对map的内容做了改变，则这些改变对于调用者是可见的。

1
2
3
4
5
6
7

attended := map[string]bool{
    "Ann": true,
    "Joe": true,
    ...}

if attended[person] { // will be false if person is not in the map
    fmt.Println(person, "was at the meeting")}

• 如果只测试是否在map中存在，而不关心实际的值，你可以将通常使用变量的地方换成空白标识符（_）

1

_, present := timeZone[tz]

• 要删除一个map项，使用delete内建函数，其参数为map和要删除的key。即使key已经不在map中，这样做也是安全的。

1

delete(timeZone, "PDT")  // Now on Standard Time

• map不太好判断是否存在某个key，如果key不存在返回的对应类型的零值，如果已有key的value恰好为零值会导致误判

打印输出

• Go中的格式化打印使用了与C中printf家族类似的风格，不过更加丰富和通用。这些函数位于fmt程序包中，并具有大写的名字：fmt.Printf，fmt.Fprintf，fmt.Sprintf等等。字符串函数（Sprintf等）返回一个字符串，而不是填充到提供的缓冲里。
• 你不需要提供一个格式串。对每个Printf，Fprintf和Sprintf，都有另外一对相应的函数，例如Print和Println。这些函数不接受格式串，而是为每个参数生成一个缺省的格式。Println版本还会在参数之间插入一个空格，并添加一个换行，而Print版本只有当两边的操作数都不是字符串的时候才增加一个空格。在这个例子中，每一行都会产生相同的输出。
• 格式化打印函数fmt.Fprint等，接受的第一个参数为任何一个实现了io.Writer接口的对象；变量os.Stdout和os.Stderr是常见的实例。
• 如果只是想要缺省的转换，像十进制整数，你可以使用 通用格式%v（代表“value”）；这正是Print和Println所产生的结果。而且，这个格式可以打印任意的的值，甚至是数组，切片，结构体和map。
• 当打印一个结构体时，带修饰的格式 %+v会将结构体的域使用它们的名字进行注解，对于任意的值，格式%#v会按照完整的Go语法打印出该值。
• 还可以通过 %q 来实现带引号的字符串格式，用于类型为 string或[]byte 的值。格式 %#q 将尽可能的使用反引号。（格式%q还用于整数和符文，产生一个带单引号的符文常量。）
• %x 用于字符串，字节数组和字节切片，以及整数，生成一个 长的十六进制字符串，并且如果在格式中 有一个空格（% x），其将会在 字节中插入空格。
• 不要在Sprintf里面调用接收者的String方法，否则会造成无穷递归，如下。只有%s匹配才会调用MyString的String方法

1
2
3
4
5
6

type MyString string

func (m MyString) String() string {
    return fmt.Sprintf("MyString=%s", m) // Error: will recur forever.
//    return fmt.Sprintf("MyString=%s", string(m)) // OK: note conversion.
}

• 另一种打印技术，是将一个打印程序的参数直接传递给另一个这样的程序。Printf的签名使用了类型…interface{}作为最后一个参数，来指定在格式之后可以出现任意数目的（任意类型的）参数。

append内建函数:

• 其中T为任意给定类型的占位符。你在Go中是无法写出一个类型T由调用者来确定的函数。这就是为什么append是内建的：它需要编译器的支持。append所做的事情是将元素添加到切片的结尾，并返回结果。

1
2
3
4
5

func append(slice []T, elements ...T) []T

x := []int{1,2,3}
x = append(x, 4, 5, 6)
fmt.Println(x)

• 如果想要在append中把一个slice添加到另一个slice要怎么做？在调用点使用 “…”，

1
2
3
4

x := []int{1,2,3}
y := []int{4,5,6}
x = append(x, y...)
fmt.Println(x)

• 可以看出 “…” 的作用是，把一个slice转为对应的type，作为一个参数列表进行传递