1.使用Spark读取并分析二进制文件
2.SPARK-38864 - Spark支持unpivot源码分析
3.为什么Spark发展不如Hadoop
4.面试 | 你真的码行了解count(*)和count(1)嘛？
5.重点 | Spark的并行度如何设置？

使用Spark读取并分析二进制文件

       客户希望通过Spark来分析二进制文件中0和1的数量及占比。针对目录下的码行每个文件进行单独分析，并将结果保存为日志文件，码行内容包括0和1字符的码行数量与占比。如果值换算为二进制不足八位，码行需在左侧填充0。码行快货运源码

       在Linux下查看二进制文件内容，码行命令为“-c 1 显示1列1个字符，码行-b 显示二进制”。码行

       使用Python版本的码行代码，核心逻辑集中在“analysis_file_content”方法中。码行

       Python脚本为命令行运行，码行无需编译。码行运行前需安装pyspark。码行在Linux环境下，码行使用pip安装时，可能遇到连接超时导致下载失败的问题，解决方法是修改连接超时值，在`~/.pip/pip.conf`中增加相关配置。odflv旧版源码安装py4j时，如果安装失败，通过执行特定安装命令，确保pyspark成功安装。

       分析结果中包含中文时，需在代码文件首行添加`# -*- coding: utf-8 -*-`声明。SparkConf初始化出现问题时，需确保传入正确参数，避免将conf误传为master参数。处理sys.argv参数时，需注意argv是一个list，其长度通过`len()`方法获取，第一个参数是python脚本文件路径，第二个参数是目标文件路径。

       在Python 2.7中，整数参与除法会得到去掉小数的结果。为解决此问题，导入`from __future__ import division`模块。git status源码在Scala版本中编译并打包生成的jar文件，通过`spark-submit`命令运行，传入参数为要分析的文件目录，格式为`file:///或hdfs://`。

       在Scala中，Byte类型为8位有符号补码整数，数值区间为-至。读取Byte数据时，即使二进制值为，其值为-1而非，因补码原则。对于-1转换为二进制字符串时，实际得到的是。针对八位的二进制数值，可编写方法将其从Byte类型转为Short类型，再使用`toBinaryString()`方法转换为二进制字符串。对于不足八位的二进制数值，可利用`String.format`进行格式化。369棋牌源码

SPARK- - Spark支持unpivot源码分析

       unpivot是数据库系统中用于列转行的内置函数，如SQL SERVER, Oracle等。以数据集tb1为例，每个数字代表某个人在某个学科的成绩。若要将此表扩展为三元组，可使用union实现。但随列数增加，SQL语句变长。许多SQL引擎提供内置函数unpivot简化此过程。unpivot使用时需指定保留列、进行转行的列、新列名及值列名。

       SPARK从SPARK-版本开始支持DataSet的unpivot函数，逐步扩展至pyspark与SQL。在Dataset API中，ids为要保留的Column数组，Column类提供了从String构造Column的隐式转换，方便使用。福利帝国源码利用此API，可通过unpivot函数将数据集转换为所需的三元组。values表示转行列，variableColumnName为新列名，valueColumnName为值列名。

       Analyser阶段解析unpivot算子，将逻辑执行计划转化为物理执行计划。当用户开启hive catalog，SPARK SQL根据表名和metastore URL查找表元数据，转化为Hive相关逻辑执行计划。物理执行计划如BroadcastHashJoinExec，表示具体的执行策略。规则ResolveUnpivot将包含unpivot的算子转换为Expand算子，在物理执行计划阶段执行。此转换由开发者自定义规则完成，通过遍历逻辑执行计划树，根据节点类型及状态进行不同处理。

       unpivot函数实现过程中，首先将原始数据集投影为包含ids、variableColumnName、valueColumnName的列，实现语义转换。随后，通过map函数处理values列，构建新的行数据，最终返回Expand算子。在物理执行计划阶段，Expand算子将数据转换为所需形式，实现unpivot功能。

       综上所述，SPARK内置函数unpivot的实现通过解析列参数，组装Expand算子完成，为用户提供简便的列转行功能。通过理解此过程，可深入掌握SPARK SQL的开发原理与内在机制。

为什么Spark发展不如Hadoop

       Spark是一个基于RAM计算的开源码ComputerCluster运算系统，目的是更快速地进行数据分析。Spark早期的核心部分代码只有3万行。Spark提供了与HadoopMap/Reduce相似的分散式运算框架，但基于RAM和优化设计，因此在交换式数据分析和datamining的Workload中表现不错。

       è¿›å…¥å¹´ä»¥åŽï¼ŒSpark开源码生态系统大幅增长，已成为大数据范畴最活跃的开源码项目之一。Spark之所以有如此多的关注，原因主要是因为Spark具有的高性能、高灵活性、与Hadoop生态系统完美融合等三方面的特点。

       é¦–先，Spark对分散的数据集进行抽样，创新地提出RDD(ResilientDistributedDataset)的概念，所有的统计分析任务被翻译成对RDD的基本操作组成的有向无环图(DAG)。RDD可以被驻留在RAM中，往后的任务可以直接读取RAM中的数据;同时分析DAG中任务之间的依赖性可以把相邻的任务合并，从而减少了大量不准确的结果输出，极大减少了HarddiskI/O，使复杂数据分析任务更高效。从这个推算，如果任务够复杂，Spark比Map/Reduce快一到两倍。

       å…¶æ¬¡ï¼ŒSpark是一个灵活的运算框架，适合做批次处理、工作流、交互式分析、流量处理等不同类型的应用，因此Spark也可以成为一个用途广泛的运算引擎，并在未来取代Map/Reduce的地位。

       æœ€åŽï¼ŒSpark可以与Hadoop生态系统的很多组件互相操作。Spark可以运行在新一代资源管理框架YARN上，它还可以读取已有并存放在Hadoop上的数据，这是个非常大的优势。

       è™½ç„¶Spark具有以上三大优点，但从目前Spark的发展和应用现状来看，Spark本身也存在很多缺陷，主要包括以下几个方面：

       â€“稳定性方面，由于代码质量问题，Spark长时间运行会经常出错，在架构方面，由于大量数据被缓存在RAM中，Java回收垃圾缓慢的情况严重，导致Spark性能不稳定，在复杂场景中SQL的性能甚至不如现有的Map/Reduce。

       â€“不能处理大数据，单独机器处理数据过大，或者由于数据出现问题导致中间结果超过RAM的大小时，常常出现RAM空间不足或无法得出结果。然而，Map/Reduce运算框架可以处理大数据，在这方面，Spark不如Map/Reduce运算框架有效。

       â€“不能支持复杂的SQL统计;目前Spark支持的SQL语法完整程度还不能应用在复杂数据分析中。在可管理性方面，SparkYARN的结合不完善，这就为使用过程中埋下隐忧，容易出现各种难题。

       è™½ç„¶Spark活跃在Cloudera、MapR、Hortonworks等众多知名大数据公司，但是如果Spark本身的缺陷得不到及时处理，将会严重影响Spark的普及和发展。

面试 | 你真的了解count(*)和count(1)嘛？

       在数据处理领域，SQL中的聚合函数count(*)和count(1)常被用于统计行数。然而，你是否真正了解这两者在Spark SQL环境下的行为和性能？本文基于Spark 3.2版本，揭示了count(*)与count(1)在功能与效率上的等价性。

       首先，给出在Spark SQL环境中，count(*)和count(1)在逻辑执行计划和最终结果方面表现一致。通过案例展示，我们可以看到当执行count(*)时，其在生成逻辑执行计划阶段即被转换为等效的count(1)操作。

       深入源码分析，我们可以发现处理count(*)与count(1)的逻辑在AstBuilder类的visitFunctionCall方法中被实现。在该方法中，处理函数节点的代码进行了优化，以高效判断表达式是否为null，进而节省计算资源。

       具体而言，count(*)功能如下：

       计算检索到的行总数，包括包含null的行。

       对于count(expr[, expr...])和count(DISTINCT expr[, expr...])，它们分别根据提供的表达式是否均为非空或唯一且非空来统计行数。

       在判断expression是否为null时，代码优先从expression的nullable属性进行判断，如果该属性无法提供明确结果，再通过isnull函数获取具体值是否为null的信息。这种策略有助于在一定程度上减少不必要的计算。

       为帮助读者更全面地理解Spark SQL的count函数，以下是推荐阅读的内容：

       澄清 | snappy压缩到底支持不支持split? 为啥？

       以后的事谁也说不准

       转型数仓开发该怎么学

       大数据开发轻量级入门方案

       OLAP | 基础知识梳理

       Flink系列 - 实时数仓之数据入ElasticSearch实战

       Flink系列 - 实时数仓之FlinkCDC实现动态分流实战

重点 | Spark的并行度如何设置？

       Spark并行度设置关键在于理解资源与数据并行度的概念。

       资源的并行度主要由节点数量和CPU核心数决定，而数据的并行度则取决于任务数据量和分区大小。任务在Spark中分为两类：Map任务和Reduce（Shuffle）任务。

       Task数量由多种因素决定，包括资源总核心数、spark.default.parallelism参数、spark.sql.shuffle.partitions参数、数据源类型、Shuffle方法的第二个参数、repartition的数目等。Task数量过多导致资源浪费，过少则形成资源瓶颈。官方建议Task数量为资源总核心数的2-3倍为佳，这样可以充分利用资源，避免部分资源闲置。

       设置Task数量时，应考虑Task执行时间的差异，以避免资源浪费。如果Task数量与CPU核心数相同，可能存在资源闲置的情况。设置2~3倍的Task数量，可以确保一旦一个Task完成，下一个Task就能迅速启动，减少资源等待时间。

       Spark并行度参数设置时，参数spark.default.parallelism可能没有默认值，但在Shuffle过程中起作用。合理设置并行度参数，可以最大化利用集群资源，提高任务执行效率。

       在实际项目中，通过数据量调整资源配置是常见的做法。设置Task数量时，应考虑数据规模，合理匹配资源，确保并行度设置既充分利用资源又不过度浪费。