跳转至

13 Spark SQL:让我们从“小汽车摇号分析”开始

你好,我是吴磊。

在开篇词我们提出“入门Spark需要三步走”,到目前为止,我们携手并肩跨越了前面两步,首先恭喜你学到这里!熟练掌握了Spark常用算子与核心原理以后,你已经可以轻松应对大部分数据处理需求了。

不过,数据处理毕竟是比较基础的数据应用场景,就像赛车有着不同的驾驶场景,想成为Spark的资深赛车手,我们还要走出第三步——学习Spark计算子框架。只有完成这一步,我们才能掌握Spark SQL,Structured Streaming和Spark MLlib的常规开发方法,游刃有余地应对不同的数据应用场景,如数据分析、流计算和机器学习,等等。

图片

那这么多子框架,从哪里入手比较好呢?在所有的子框架中,Spark SQL是代码量最多、Spark社区投入最大、应用范围最广、影响力最深远的那个。就子框架的学习来说,我们自然要从Spark SQL开始。

今天我们从一个例子入手,在实战中带你熟悉数据分析开发的思路和实现步骤。有了对Spark SQL的直观体验,我们后面几讲还会深入探讨Spark SQL的用法、特性与优势,让你逐步掌握Spark SQL的全貌。

业务需求

今天我们要讲的小例子,来自于北京市小汽车摇号。我们知道,为了限制机动车保有量,从2011年开始,北京市政府推出了小汽车摇号政策。随着摇号进程的推进,在2016年,为了照顾那些长时间没有摇中号码牌的“准司机”,摇号政策又推出了“倍率”制度。

所谓倍率制度,它指的是,结合参与摇号次数,为每个人赋予不同的倍率系数。有了倍率加持,大家的中签率就由原来整齐划一的基础概率,变为“基础概率 * 倍率系数”。参与摇号的次数越多,倍率系数越大,中签率也会相应得到提高。

不过,身边无数的“准司机”总是跟我说,其实倍率这玩意没什么用,背了8倍、10倍的倍率,照样摇不上!那么今天这一讲,咱们就来借着学习Spark SQL的机会,用数据来为这些还没摸过车的“老司机”答疑解惑,帮他们定量地分析一下,倍率与中签率之间,到底有没有关系?

准备工作

巧妇难为无米之炊,既然是做数据分析,那咱们得先有数据才行。我这边为你准备了2011年到2019年北京市小汽车的摇号数据,你可以通过这个地址,从网盘进行下载,提取码为ajs6。

这份数据的文件名是“2011-2019 小汽车摇号数据.tar.gz”,解压之后的目录结构如下图所示。

可以看到,根目录下有apply和lucky两个子目录,apply目录的内容是 2011-2019 年各个批次参与摇号的申请号码,而lucky目录包含的是各个批次中签的申请号码。为了叙述方便,我们把参与过摇号的人叫“申请者”,把中签的人叫“中签者”。apply和lucky的下一级子目录是各个摇号批次,而摇号批次目录下包含的是Parquet格式的数据文件。

图片

数据下载、解压完成之后,接下来,我们再来准备运行环境。

咱们的小例子比较轻量,Scala版本的代码实现不会超过20行,再者摇号数据体量很小,解压之后的Parquet文件总大小也不超过4G。

选择这样的例子也是为了轻装上阵,避免你因为硬件限制而难以实验。想要把用于分析倍率的应用跑起来,你在笔记本或是PC上,通过启动本地spark-shell环境就可以。不过,如果条件允许的话,我还是鼓励你搭建分布式的物理集群。关于分布式集群的搭建细节,你可以参考第4讲

好啦,准备好数据与运行环境之后,接下来,我们就可以步入正题,去开发探索倍率与中签率关系的数据分析应用啦。

数据探索

不过,先别忙着直接上手数据分析。在此之前,我们先要对数据模式(Data Schema)有最基本的认知,也就是源数据都有哪些字段,这些字段的类型和含义分别是什么,这一步就是我们常说的数据探索。

数据探索的思路是这样的:首先,我们使用SparkSession的read API读取源数据、创建DataFrame。然后,通过调用DataFrame的show方法,我们就可以轻松获取源数据的样本数据,从而完成数据的初步探索,代码如下所示。

import org.apache.spark.sql.DataFrame
 
val rootPath: String = _
// 申请者数据
val hdfs_path_apply: String = s"${rootPath}/apply"
// spark是spark-shell中默认的SparkSession实例
// 通过read API读取源文件
val applyNumbersDF: DataFrame = spark.read.parquet(hdfs_path_apply)
// 数据打印
applyNumbersDF.show
 
// 中签者数据
val hdfs_path_lucky: String = s"${rootPath}/lucky"
// 通过read API读取源文件
val luckyDogsDF: DataFrame = spark.read.parquet(hdfs_path_lucky)
// 数据打印
luckyDogsDF.show

看到这里,想必你已经眉头紧锁:“SparkSession?DataFrame?这些都是什么鬼?你好像压根儿也没有提到过这些概念呀!”别着急,对于这些关键概念,我们在后续的课程中都会陆续展开,今天这一讲,咱们先来“知其然”,“知其所以然”的部分咱们放到后面去讲。

对于SparkSession,你可以把它理解为是SparkContext的进阶版,是Spark(2.0版本以后)新一代的开发入口。SparkContext通过textFile API把源数据转换为RDD,而SparkSession通过read API把源数据转换为DataFrame。

而DataFrame,你可以把它看作是一种特殊的RDD。RDD我们已经很熟悉了,现在就把DataFrame跟RDD做个对比,让你先对DataFrame有个感性认识。

先从功能分析,与RDD一样,DataFrame也用来封装分布式数据集,它也有数据分区的概念,也是通过算子来实现不同DataFrame之间的转换,只不过DataFrame采用了一套与RDD算子不同的独立算子集。

再者,在数据内容方面,与RDD不同,DataFrame是一种带Schema的分布式数据集,因此,你可以简单地把DataFrame看作是数据库中的一张二维表。

最后,DataFrame背后的计算引擎是Spark SQL,而RDD的计算引擎是Spark Core,这一点至关重要。不过,关于计算引擎之间的差异,我们留到下一讲再去展开。

好啦,言归正传。简单了解了SparkSession与DataFrame的概念之后,我们继续来看数据探索。

把上述代码丢进spark-shell之后,分别在applyNumbersDF和luckyDogsDF这两个DataFrame之上调用show函数,我们就可以得到样本数据。可以看到,“这两张表”的Schema是一样的,它们都包含两个字段,一个是String类型的carNum,另一个是类型为Int的batchNum。

图片

其中,carNum的含义是申请号码、或是中签号码,而batchNum则代表摇号批次,比如201906表示2019年的最后一批摇号,201401表示2014年的第一次摇号。

好啦,进行到这里,初步的数据探索工作就告一段落了。

业务需求实现

完成初步的数据探索之后,我们就可以结合数据特点(比如两张表的Schema完全一致,但数据内容的范畴不同),来实现最开始的业务需求:计算中签率与倍率之间的量化关系。

首先,既然是要量化中签率与倍率之间的关系,我们只需要关注那些中签者(lucky目录下的数据)的倍率变化就好了。而倍率的计算,要依赖apply目录下的摇号数据。因此,要做到仅关注中签者的倍率,我们就必须要使用数据关联这个在数据分析领域中最常见的操作。此外,由于倍率制度自2016年才开始推出,所以我们只需要访问2016年以后的数据即可。

基于以上这些分析,我们先把数据过滤与数据关联的代码写出来,如下所示。

// 过滤2016年以后的中签数据,且仅抽取中签号码carNum字段
val filteredLuckyDogs: DataFrame = luckyDogsDF.filter(col("batchNum") >= "201601").select("carNum")
 
// 摇号数据与中签数据做内关联,Join Key为中签号码carNum
val jointDF: DataFrame = applyNumbersDF.join(filteredLuckyDogs, Seq("carNum"), "inner")

在上面的代码中,我们使用filter算子对luckyDogsDF做过滤,然后使用select算子提取carNum字段。

紧接着,我们在applyNumbersDF之上调用join算子,从而完成两个DataFrame的数据关联。join算子有3个参数,你可以对照前面代码的第5行来理解,这里第一个参数用于指定需要关联的DataFrame,第二个参数代表Join Key,也就是依据哪些字段做关联,而第三个参数指定的是关联形式,比如inner表示内关联,left表示左关联,等等。

做完数据关联之后,接下来,我们再来说一说,倍率应该怎么统计。对于倍率这个数值,官方的实现略显粗暴,如果去观察 apply 目录下 2016 年以后各个批次的文件,你就会发现,所谓的倍率,实际上就是申请号码的副本数量。

比如说,我的倍率是8,那么在各个批次的摇号文件中,我的申请号码就会出现8次。是不是很粗暴?因此,要统计某个申请号码的倍率,我们只需要统计它在批次文件中出现的次数就可以达到目的。

按照批次、申请号码做统计计数,是不是有种熟悉的感觉?没错,这不就是我们之前学过的Word Count吗?它本质上其实就是一个分组计数的过程。不过,这一次,咱们不再使用reduceByKey这个RDD算子了,而是使用DataFrame的那套算子来实现,我们先来看代码。

val multipliers: DataFrame = jointDF.groupBy(col("batchNum"),col("carNum"))
.agg(count(lit(1)).alias("multiplier"))

分组计数

对照代码我给你分析下思路,我们先是用groupBy算子来按照摇号批次和申请号码做分组,然后通过agg和count算子把(batchNum,carNum)出现的次数,作为carNum在摇号批次batchNum中的倍率,并使用alias算子把倍率重命名为“multiplier”。

这么说可能有点绕,我们可以通过在multipliers之上调用show函数,来直观地观察这一步的计算结果。为了方便说明,我用表格的形式来进行示意。

图片

可以看到,同一个申请号码,在不同批次中的倍率是不一样的。就像我们之前说的,随着摇号的次数增加,倍率也会跟着提升。不过,这里咱们要研究的是倍率与中签率的关系,所以只需要关心中签者是在多大的倍率下中签的就行。因此,对于同一个申请号码,我们只需要保留其中最大的倍率就可以了。

需要说明的是,取最大倍率的做法,会把倍率的统计基数变小,从而引入幸存者偏差。更严谨的做法,应该把中签者过往的倍率也都统计在内,这样倍率的基数才是准确的。不过呢,结合实验,幸存者偏差并不影响“倍率与中签率是否有直接关系”这一结论。因此,咱们不妨采用取最大倍率这种更加简便的做法。毕竟,学习Spark SQL,才是咱们的首要目标。

为此,我们需要“抹去”batchNum这个维度,按照carNum对multipliers做分组,并提取倍率的最大值,代码如下所示。

val uniqueMultipliers: DataFrame = multipliers.groupBy("carNum")
.agg(max("multiplier").alias("multiplier"))

分组聚合的方法跟前面差不多,我们还是先用groupBy做分组,不过这次仅用carNum一个字段做分组,然后使用agg和max算子来保留倍率最大值。经过这一步的计算之后,我们就得到了每个申请号码在中签之前的倍率系数:

图片

可以看到,uniqueMultipliers这个DataFrame仅包含申请号码carNum和倍率multiplier这两个字段,且carNum字段不存在重复值,也就是说,在这份数据集中,一个申请号码,只有一个最大倍率与之对应。

好啦,到此为止,我们拿到了每一个中签者,在中签之前的倍率系数。接下来,结合这份数据,我们就可以统计倍率本身的分布情况。

具体来说,我们想知道的是,不同倍率之下的人数分布是什么样子的。换句话说,这一次,我们要按照倍率来对数据做分组,然后计算不同倍率下的统计计数。不用说,这次咱们还是得仰仗groupBy和agg这两个算子,代码如下所示。

val result: DataFrame = uniqueMultipliers.groupBy("multiplier")
.agg(count(lit(1)).alias("cnt"))
.orderBy("multiplier")
 
result.collect

在最后一步,我们依然使用groupBy和agg算子如法炮制,得到按照倍率统计的人数分布之后,我们通过collect算子来收集计算结果,并同时触发上述的所有代码从头至尾交付执行。

计算结果result包含两个字段,一个是倍率,一个是持有该倍率的统计人数。如果把result结果数据做成柱状图的话,我们可以更加直观地观察到中签率与倍率之间的关系,如下图所示。

图片

不难发现,不同倍率下的中签者人数,呈现出正态分布。也即是说,对于一个申请者来说,他/她有幸摇中的概率,并不会随着倍率的增加而线性增长。用身边那些“老司机”的话说,中签这件事,确实跟倍率的关系不大。

重点回顾

今天这一讲,我们一起动手,开发了“倍率的统计分布”这个数据分析应用,并解答了中签率与倍率之间是否存在关联关系这一难题。

尽管在实现的过程中,我们遇到了一些新概念和新的算子,但你不必担心,更不必着急。今天这节课,你只需要对Spark SQL框架下的应用开发有一个感性的认识就可以了。

在Spark SQL的开发框架下,我们通常是通过SparkSession的read API从源数据创建DataFrame。然后,以DataFrame为入口,在DataFrame之上调用各式各样的转换算子,如agg、groupBy、select、filter等等,对DataFrame进行转换,进而完成相应的数据分析。

为了后续试验方便,我把今天涉及的代码片段整理到了一起,你可以把它们丢进spark-shell去运行,观察每个环节的计算结果,体会不同算子的计算逻辑与执行结果之间的关系。加油,祝你好运!

import org.apache.spark.sql.DataFrame
 
val rootPath: String = _
// 申请者数据
val hdfs_path_apply: String = s"${rootPath}/apply"
// spark是spark-shell中默认的SparkSession实例
// 通过read API读取源文件
val applyNumbersDF: DataFrame = spark.read.parquet(hdfs_path_apply)
 
// 中签者数据
val hdfs_path_lucky: String = s"${rootPath}/lucky"
// 通过read API读取源文件
val luckyDogsDF: DataFrame = spark.read.parquet(hdfs_path_lucky)
 
// 过滤2016年以后的中签数据,且仅抽取中签号码carNum字段
val filteredLuckyDogs: DataFrame = luckyDogsDF.filter(col("batchNum") >= "201601").select("carNum")
 
// 摇号数据与中签数据做内关联,Join Key为中签号码carNum
val jointDF: DataFrame = applyNumbersDF.join(filteredLuckyDogs, Seq("carNum"), "inner")
 
// 以batchNum、carNum做分组,统计倍率系数
val multipliers: DataFrame = jointDF.groupBy(col("batchNum"),col("carNum"))
.agg(count(lit(1)).alias("multiplier"))
 
// 以carNum做分组,保留最大的倍率系数
val uniqueMultipliers: DataFrame = multipliers.groupBy("carNum")
.agg(max("multiplier").alias("multiplier"))
 
// 以multiplier倍率做分组,统计人数
val result: DataFrame = uniqueMultipliers.groupBy("multiplier")
.agg(count(lit(1)).alias("cnt"))
.orderBy("multiplier")
 
result.collect

每课一练

1.脑洞时间:你觉得汽车摇号的倍率制度应该怎样设计,才是最合理的?

2.请在你的Spark环境中把代码运行起来,并确认执行结果是否与result一致。

欢迎你在留言区跟我交流互动,也推荐你把这一讲的内容分享给更多的朋友、同事。我们下一讲见!

精选留言(15)
  • qinsi 👍(16) 💬(3)

    spark无关。讨论下摇号。 评论区有匿名读者质疑文中的结论。这里尝试换个角度代入具体的数字分析下。 简单起见,假设每轮摇号有1000人中签,并且倍率和轮次一致,即第一轮大家都是1倍,第一轮没中的人在第二轮变为2倍,第二轮又没中的人到了第三轮就变成3倍,依次类推。 先看第一轮的1000个中签者,显然他们的倍率都是1,没有其他倍率的中签者,记为: [1000, 0, 0, ...] 再看第二轮的1000个中签者。由于新加入的申请者倍率为1,第一轮未中的人倍率为2。按照倍率摇号的话,期望的结果就是,倍率为2的中签人数是倍率为1的中签人数的2倍,记为: [333, 667, 0, 0, ...] 以此类推,第三轮就是: [167, 333, 500, 0, 0, ...] 尝试摇个10轮,可以得到下表: [1000, 0, 0, ...] [333, 667, 0, 0, ...] [167, 333, 500, 0, 0, ...] [100, 200, 300, 400, 0, 0, ...] [67, 133, 200, 267, 333, 0, 0, ...] [48, 95, 143, 190, 238, 286, 0, 0, ...] [36, 71, 107, 143, 179, 214, 250, 0, 0, ...] [28, 56, 83, 111, 139, 167, 194, 222, 0, 0, ...] [22, 44, 67, 89, 111, 133, 156, 178, 200, 0, 0, ...] [18, 36, 55, 73, 91, 109, 127, 145, 164, 182, 0, 0, ...] 可以看到在每一轮的中签者中,确实是倍率越高中签的人数越多。 而文中的统计方法,相当于把这张表按列求和: [1819, 1635, 1455, 1273, 1091, 909, 727, 545, 364, 182, 0, 0, ...] 可以看到这是一条单调递减的曲线。然而却不能像文中一样得出“中签率没有随着倍率增加”的结论。高倍率的中签人数比低倍率的人数少,是因为能达到高倍率的人本身就少。比如上面例子中,10轮过后10倍率的中签者只有182人,是因为前9轮没有人能达到10倍率。相比之下,在第一轮就有1000个1倍率的人中签。 至于文中配图为什么会是一条类似钟型的曲线,猜测可能第一次引入倍率摇号的时候,就已经给不同的人分配不同的倍率了,而不是大家一开始都是1倍率。在上面的例子中,如果只对后5轮求和,可以得到: [152, 302, 455, 606, 758, 909, 727, 545, 364, 182] 这样就和文中的配图比较接近了。 所以结论就是要验证中签率和倍率的关系,不能按照倍率去累加中签人数,而是要看单次摇号中不同倍率的中签者的分布。

    2021-10-12

  • Alvin-L 👍(6) 💬(2)

    ``` import os from pyspark import SparkContext, SparkConf from pyspark.sql.session import SparkSession from pyspark.sql.functions import first, collect_list, mean, count, max import matplotlib.pyplot as plt def plot(res): x = [x["multiplier"] for x in res] y = [y["cnt"] for y in res] plt.figure(figsize=(8, 5), dpi=100) plt.xlabel('倍率') plt.ylabel('人数') plt.rcParams['font.sans-serif']=['SimHei'] plt.rcParams['axes.unicode_minus']=False plt.bar(x, y, width=0.5) plt.xticks(x) plt.show() # py文件就在项目的根目录下 rootPath = os.path.split(os.path.realpath(__file__))[0] conf = SparkConf() conf.set('spark.executor.memory', '4g') conf.set('spark.driver.memory', '8g') conf.set("spark.executor.cores", '4') conf.set('spark.cores.max', 16) conf.set('spark.local.dir', rootPath) spark = SparkSession(SparkContext(conf=conf)) # 申请者数据 # Windows环境 # 注意点1:增加 option("basePath", rootPath) 选项 # 注意点2:路径 hdfs_path_apply 需要追加 /*/*.parquet hdfs_path_apply = rootPath + "/apply" applyNumbersDF = spark.read.option("basePath", rootPath).parquet( hdfs_path_apply + "/*/*.parquet" ) # 中签者数据 hdfs_path_lucky = rootPath + "/lucky" luckyDogsDF = spark.read.option("basePath", rootPath).parquet( hdfs_path_lucky + "/*/*.parquet" ) # 过滤2016年以后的中签数据,且仅抽取中签号码carNum字段 filteredLuckyDogs = ( luckyDogsDF .filter(luckyDogsDF["batchNum"] >= "201601") .select("carNum") ) # 摇号数据与中签数据做内关联,Join Key为中签号码carNum jointDF = applyNumbersDF.join(filteredLuckyDogs, "carNum", "inner") # 以batchNum、carNum做分组,统计倍率系数 multipliers = ( jointDF .groupBy(["batchNum", "carNum"]) .agg(count("batchNum").alias("multiplier")) ) # 以carNum做分组,保留最大的倍率系数 uniqueMultipliers = ( multipliers .groupBy("carNum") .agg(max("multiplier").alias("multiplier")) ) # 以multiplier倍率做分组,统计人数 result = ( uniqueMultipliers .groupBy("multiplier") .agg(count("carNum").alias("cnt")) .orderBy("multiplier") ) result.show(40) res = result.collect() # 画图 plot(res) ```

    2021-10-26

  • 东围居士 👍(2) 💬(2)

    补一个完整的 spark 代码(windows环境): package spark.basic import org.apache.spark.sql.functions.{col,count, lit, max} import org.apache.spark.sql.{DataFrame, SparkSession} object Chapter13 { def main(args: Array[String]): Unit = { val spark: SparkSession = SparkSession.builder().master("local[*]").appName("Chapter13").getOrCreate() import spark.implicits._ val rootPath: String = "E:\\temp\\yaohao_home\\yaohao" // 申请者数据 val hdfs_path_apply: String = s"${rootPath}/apply" // spark是spark-shell中默认的SparkSession实例 // 通过read API读取源文件 val applyNumbersDF: DataFrame = spark.read.option("basePath", rootPath).parquet(hdfs_path_apply + "/*/*.parquet") // 中签者数据 val hdfs_path_lucky: String = s"${rootPath}/lucky" // 通过read API读取源文件 val luckyDogsDF: DataFrame = spark.read.option("basePath", rootPath).parquet(hdfs_path_lucky + "/*/*.parquet") // 过滤2016年以后的中签数据,且仅抽取中签号码carNum字段 val filteredLuckyDogs: DataFrame = luckyDogsDF.filter(col("batchNum") >= "201601").select("carNum") // 摇号数据与中签数据做内关联,Join Key为中签号码carNum val jointDF: DataFrame = applyNumbersDF.join(filteredLuckyDogs, Seq("carNum"), "inner") // 以batchNum、carNum做分组,统计倍率系数 val multipliers: DataFrame = jointDF.groupBy(col("batchNum"),col("carNum")) .agg(count(lit(1)).alias("multiplier")) // 以carNum做分组,保留最大的倍率系数 val uniqueMultipliers: DataFrame = multipliers.groupBy("carNum") .agg(max("multiplier").alias("multiplier")) // 以multiplier倍率做分组,统计人数 val result: DataFrame = uniqueMultipliers.groupBy("multiplier") .agg(count(lit(1)).alias("cnt")) .orderBy("multiplier") result.collect result.show() } }

    2021-11-12

  • 火炎焱燚 👍(2) 💬(2)

    对应的python代码为: # 在notebook上运行时,加上下面的配置 from pyspark import SparkContext, SparkConf from pyspark.sql.session import SparkSession sc_conf = SparkConf() # spark参数配置 # sc_conf.setMaster() # sc_conf.setAppName('my-app') sc_conf.set('spark.executor.memory', '2g') sc_conf.set('spark.driver.memory', '4g') sc_conf.set("spark.executor.cores", '2') sc_conf.set('spark.cores.max', 20) sc = SparkContext(conf=sc_conf) # 加载数据,转换成dataframe rootPath='~~/RawData' hdfs_path_apply=rootPath+'/apply' spark = SparkSession(sc) applyNumbersDF=spark.read.parquet(hdfs_path_apply) # applyNumbersDF.show() # 打印出前几行数据,查看数据结构 hdfs_path_lucky=rootPath+'/lucky' luckyDogsDF=spark.read.parquet(hdfs_path_lucky) # luckyDogsDF.show() filteredLuckyDogs=luckyDogsDF.filter(luckyDogsDF['batchNum']>='201601').select('carNum') jointDF=applyNumbersDF.join(filteredLuckyDogs,'carNum','inner') # join函数消耗内存较大,容易出现OOM错误,如果出错,要将spark.driver.memory调大 # jointDF.show() # 打印出join之后的df部分数据 # 进行多种groupBy操作 from pyspark.sql import functions as f multipliers=jointDF.groupBy(['batchNum','carNum']).agg(f.count('batchNum').alias("multiplier")) # multipliers.show() uniqueMultipliers=multipliers.groupBy('carNum').agg(f.max('multiplier').alias('multiplier')) # uniqueMultipliers.show() result=uniqueMultipliers.groupBy('multiplier').agg(f.count('carNum').alias('cnt')).orderBy('multiplier') result2=result.collect() # 绘图 import matplotlib.pyplot as plt x=[i['multiplier'] for i in result2] y=[i['cnt'] for i in result2] plt.bar(x,y)

    2021-10-23

  • Geek_d447af 👍(1) 💬(2)

    文章里的代码需要在 Hadoop 环境才能跑起来,spark 本身不支持解析 parquet 文件

    2021-10-09

  • lightning_女巫 👍(0) 💬(1)

    我在本地跑这个代码碰到了如下错误,请问如何解决? 22/01/28 15:13:22 ERROR BypassMergeSortShuffleWriter: Error while deleting file /private/var/folders/hk/7j9sqdtn55j3cq_gv5qvp5pm39d49n/T/blockmgr-88ef94e9-943a-4971-a3a8-33d25949886f/1a/temp_shuffle_e0e163fb-852c-4298-b08e-dc4989277ab3 22/01/28 15:13:22 ERROR DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file /private/var/folders/hk/7j9sqdtn55j3cq_gv5qvp5pm39d49n/T/blockmgr-88ef94e9-943a-4971-a3a8-33d25949886f/08/temp_shuffle_6c160c23-3395-445f-be03-b29a375e1139 java.io.FileNotFoundException: /private/var/folders/hk/7j9sqdtn55j3cq_gv5qvp5pm39d49n/T/blockmgr-88ef94e9-943a-4971-a3a8-33d25949886f/08/temp_shuffle_6c160c23-3395-445f-be03-b29a375e1139 (No such file or directory) at java.io.FileOutputStream.open0(Native Method) at java.io.FileOutputStream.open(FileOutputStream.java:270) at java.io.FileOutputStream.<init>(FileOutputStream.java:213) at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose$2.apply$mcV$sp(DiskBlockObjectWriter.scala:217) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1369) at org.apache.spark.storage.DiskBlockObjectWriter.revertPartialWritesAndClose(DiskBlockObjectWriter.scala:214) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.stop(BypassMergeSortShuffleWriter.java:237) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:105) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at org.apache.spark.scheduler.Task.run(Task.scala:123) at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)

    2022-01-28

  • 东围居士 👍(0) 💬(3)

    老师,数据文件方便存一份到别的地方吗,比如马云家的网盘,或者做个种子下载什么的,百度网盘那速度真的是,我下到下午下班过周末都下不完

    2021-10-22

  • Geek_995b78 👍(0) 💬(2)

    用scala实现,lit(1)是什么意思呀

    2021-10-11

  • GAC·DU 👍(0) 💬(1)

    result具体数值: scala> result.collect res7: Array[org.apache.spark.sql.Row] = Array([1,8967], [2,19174], [3,26952], [4,29755], [5,32988], [6,34119], [7,29707], [8,26123], [9,19476], [10,9616], [11,3930], [12,1212])

    2021-10-08

  • Neo-dqy 👍(0) 💬(1)

    【.agg(count(lit(1)).alias("cnt"))】问下老师,这里count中的lit(1)是什么意思啊? 对于汽车摇号的倍率制度,如果为了优先让倍率高的人摇到号,可以把每一期的资格分多次抽取。就是说,先构建一个所有人都在里面的样本,抽部分人;再将倍率高于某个阈值的人都取出来,构建一个新的样本,再抽取部分人。(具体划分成几个样本可以按倍率的人数分布来划分)当然这样又会对新来的人不公平,所以大家还是挤地铁吧~~

    2021-10-08

  • Spoon 👍(2) 💬(0)

    Java实现 https://github.com/Spoon94/spark-practice/blob/master/src/main/java/com/spoon/spark/sql/CarNumAnalyseJob.java

    2022-04-05

  • 未来已来 👍(1) 💬(0)

    大概看了下评论,发现一次摇号一个号码会出现多次,是为了增加n次参与的人被摇到的概率。相当于在一个封闭的箱子里摇球,一个号码的球多放了几个,摇箱子后个数多的号码被抽到的概率更高(n/N,N为箱子内球的总数)

    2023-02-10

  • 翡翠小南瓜 👍(1) 💬(2)

    不懂北京的摇号规则,也没写清楚,所以是一个批次号里面,一个申请号可以有多次????

    2022-04-12

  • Each 👍(0) 💬(0)

    老師您好, 無法下載 dataset, 可以提供海外載點嗎? 謝謝.

    2024-10-15

  • 风一样 👍(0) 💬(1)

    上面的结论感觉不太正确,偏离了每个每个倍率下的基数,倍率越高中签概率肯定越大啊,对个人而言,如果我们每次参与抽签的人数大基数不变的情况下(基数1000),倍率越大,相当于往里面添加了多个样本(8倍率),本来是1/1000的概率变为了8/1000。但是考虑到每次倍率越大没中签的用户重复的次数越多,基数也会跟着变大,相对于新加入的人来说其实是有优势的,但是新增的人其实远小于已经参与过抽签的人,所以才导致了感觉上变化不大

    2022-12-01