372 lines
13 KiB
Markdown
372 lines
13 KiB
Markdown
Python版本：3.7.9
|
||
Spark版本：2.4.7
|
||
包版本：org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.7
|
||
|
||
## 程序报没有写hdfs权限，但是程序本身没有写hdfs：
|
||
|
||
|
||
|
||
|
||
1. 原因：没有设置checkpoint地址，程序尝试写入到默认地址，但在默认地址没有读写权限；
|
||
|
||
2. 解决方法：设置checkpoint地址:.option("checkpointLocation", "地址")
|
||
|
||
## 程序报错java.nio.ByteBuffer.allocate(ByteBuffer.java:334)：
|
||
|
||
```shell
|
||
java.lang.IllegalArgumentException at java.nio.ByteBuffer.allocate(ByteBuffer.java:334)
|
||
|
||
at org.apache.arrow.vector.ipc.message.MessageSerializer.readMessage(MessageSerializer.java:543)
|
||
|
||
at org.apache.arrow.vector.ipc.message.MessageChannelReader.readNext(MessageChannelReader.java:58)
|
||
|
||
at org.apache.arrow.vector.ipc.ArrowStreamReader.readSchema(ArrowStreamReader.java:132)
|
||
|
||
at org.apache.arrow.vector.ipc.ArrowReader.initialize(ArrowReader.java:181)
|
||
|
||
at org.apache.arrow.vector.ipc.ArrowReader.ensureInitialized(ArrowReader.java:172)
|
||
|
||
at org.apache.arrow.vector.ipc.ArrowReader.getVectorSchemaRoot(ArrowReader.java:65)
|
||
|
||
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:162)
|
||
|
||
at org.apache.spark.sql.execution.python.ArrowPythonRunner$$anon$1.read(ArrowPythonRunner.scala:122)
|
||
|
||
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:406)
|
||
|
||
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
|
||
|
||
at org.apache.spark.sql.execution.python.ArrowEvalPythonExec$$anon$2.<init>(ArrowEvalPythonExec.scala:98)
|
||
|
||
at org.apache.spark.sql.execution.python.ArrowEvalPythonExec.evaluate(ArrowEvalPythonExec.scala:96)
|
||
|
||
at org.apache.spark.sql.execution.python.EvalPythonExec$$anonfun$doExecute$1.apply(EvalPythonExec.scala:127)...
|
||
```
|
||
|
||
|
||
|
||
1. 错误原因：pyarrow版本不兼容导致;
|
||
|
||
|
||
2. 解决方法：
|
||
|
||
1. 将pyarrow版本降到0.15.0以下，开发中使用的是0.14.0版本
|
||
|
||
2. 在spark安装目录的conf文件夹中的spark-env.sh文件中加入export RROW_PRE_0_15_IPC_FORMAT=1
|
||
|
||
## 程序报错 Did not pass numpy.dtype object：
|
||
|
||
```shell
|
||
pyarrow.lib.ArrowTypeError: ('Did not pass numpy.dtype object', 'Conversion failed for column IN_MU_user_fee with type bool')
|
||
```
|
||
|
||
|
||
1. 错误原因：numpy版本不兼容导致；
|
||
|
||
2. 解决方法：将numpy版本降到1.20.0以下，开发中使用的是1.19.1
|
||
|
||
## 环境缺失组件：开发中遇到缺失rediscluster组件问题；
|
||
|
||
|
||
|
||
|
||
|
||
1. 解决方法(指定Python zip包)：
|
||
|
||
1. 将所需的模块从安装目录打包为Zip包；
|
||
|
||
2. 上zip包上传至服务器；
|
||
|
||
3. 用hadoop fs -mkdir hdfs:/user/faw_vhm_admin/lizhenyang/建立文件夹；
|
||
|
||
4. 用hadoop fs -put python.zip hdfs:/user/faw_vhm_admin/lizhenyang/将zip包存入上面建立的文件夹内;
|
||
|
||
5. 在提交spark任务时，使用
|
||
```shell
|
||
--archives hdfs:/user/faw_vhm_admin/lizhenyang/spark_python_test.zip --conf spark.pyspark.python=./spark_python_test.zip/python374/bin/python3.7
|
||
```
|
||
|
||
## pandas_udf函数不支持window聚合
|
||
|
||
1. 原因：在使用grouped_map模式的pandas_udf时，pandas_udf会将所有列都转为pandas的dataframe格式，这对window列是不可行的，因此引起报错
|
||
|
||
|
||
|
||
## 终端显示程序运行完毕，但是UI中显示有步骤Failed
|
||
|
||
|
||
|
||
|
||
失败原因：Stage cancelled because SparkContext was shut down
|
||
|
||
1. 原因：集群的内存不足(设置使用的内存不足，非集群物理内存不足)；
|
||
|
||
2. 解决方案：
|
||
|
||
在提交运算时，使用--driver-memory 16G --executor-memory 16G --executor-cores 12 --num-executors 12参数将可用内存、CPU核数及节点数增大；
|
||
|
||
## 在Structured steaming中实时向redis写入输入
|
||
|
||
1. 解决方法：自定义sink：
|
||
|
||
1. 定义save_to_redis函数
|
||
|
||
2. 在writestream中使用foreach(save_to_redis)，foreach函数会将每一个行分别输入save_to_redis函数，在save_to_redis函数中对输入的每一行输入存入Redis
|
||
|
||
## 在json序列化的时候遇到datetime不可序列化
|
||
|
||
1. 报错内容：TypeError: Object of type ‘datetime‘ is not JSON serializable
|
||
|
||
2. 解决方案1：将需要序列化的列中的datetime转化为字符串
|
||
```python
|
||
str(df1[df1.collectTime != np.nan]["collectTime"].values[0])
|
||
```
|
||
|
||
|
||
3. 解决方案2：重写构造json类，遇到datetime类时特殊处理
|
||
|
||
```python
|
||
from datetime import date, datetime
|
||
|
||
class ComplexEncoder(json.JSONEncoder):
|
||
|
||
def default(self, obj):
|
||
|
||
if isinstance(obj, datetime):
|
||
|
||
return obj.strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
elif isinstance(obj, date):
|
||
|
||
return obj.strftime('%Y-%m-%d')
|
||
|
||
else:
|
||
|
||
return json.JSONEncoder.default(self, obj)
|
||
```
|
||
并在json.dumps()时指定cls为ComplexEncoder
|
||
```python
|
||
import json
|
||
|
||
json.dumps(your_data, cls=ComplexEncoder)
|
||
```
|
||
|
||
## 出现极大的统计项
|
||
|
||
1. 在部分指标中，出现上亿、上千万的值，如下图
|
||
|
||
|
||
|
||
2. 根本原因：在将新增数据的dataframe与原有数据的dataframe合并时，使用了union函数，而union函数是按列的位置进行合并，新增数据和原有数据dataframe中各个列的位置不同，从而导致将新增数据中的max_speed列错误合并到了rapid_acc_times列，导致错误
|
||
|
||
|
||
|
||
|
||
|
||
|
||
3. 解决方法：使用unionbyname函数，该函数按列名进行合并，而不是列的位置；
|
||
|
||
|
||
|
||
## 在Spark SQL中将同一列的两个array量合并
|
||
|
||
1. 在统计dsm报警时间的时候，需要将同vin相同行程的dsm报警时间合并，因此需要在按vin码Group后将dsm_time_list列中的array合并；
|
||
|
||
2. 解决方法：先使用collect_list函数，再使用udf函数
|
||
|
||
[参考](https://stackoverflow.com/questions/48406304/groupby-and-concat-array-columns-pyspark/48407212](https://stackoverflow.com/questions/48406304/groupby-and-concat-array-columns-pyspark/48407212)
|
||
|
||
### python 2.x
|
||
|
||
```python
|
||
spark.version
|
||
|
||
# u'2.2.0'
|
||
|
||
from pyspark.sql import functions as F
|
||
|
||
import pyspark.sql.types as T
|
||
|
||
|
||
def fudf(val):
|
||
|
||
return reduce (lambda x, y:x+y, val)
|
||
|
||
flattenUdf = F.udf(fudf, T.ArrayType(T.IntegerType()))
|
||
|
||
df2 = df.groupBy("store").agg(F.collect_list("values"))
|
||
|
||
df2.show(truncate=False)
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
# |store| collect_list(values) |
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
# |1 |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6)]|
|
||
|
||
# |2 |[WrappedArray(2), WrappedArray(3)] |
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
|
||
df3 = df2.select("store", flattenUdf("collect_list(values)").alias("values"))
|
||
|
||
df3.show(truncate=False)
|
||
|
||
# +-----+------------------+
|
||
|
||
# |store| values |
|
||
|
||
# +-----+------------------+
|
||
|
||
# |1 |[1, 2, 3, 4, 5, 6]|
|
||
|
||
# |2 |[2, 3] |
|
||
|
||
# +-----+------------------+
|
||
|
||
```
|
||
|
||
### Python 3.x
|
||
|
||
```python
|
||
spark.version
|
||
|
||
# u'2.2.0'
|
||
|
||
from pyspark.sql import functions as F
|
||
|
||
import pyspark.sql.types as T
|
||
|
||
import functools
|
||
|
||
def fudf(val):
|
||
|
||
return functools.reduce(lambda x, y:x+y, val)
|
||
|
||
flattenUdf = F.udf(fudf, T.ArrayType(T.IntegerType()))
|
||
|
||
df2 = df.groupBy("store").agg(F.collect_list("values"))
|
||
|
||
df2.show(truncate=False)
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
# |store| collect_list(values) |
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
# |1 |[WrappedArray(1, 2, 3), WrappedArray(4, 5, 6)]|
|
||
|
||
# |2 |[WrappedArray(2), WrappedArray(3)] |
|
||
|
||
# +-----+----------------------------------------------+
|
||
|
||
df3 = df2.select("store", flattenUdf("collect_list(values)").alias("values"))
|
||
|
||
df3.show(truncate=False)
|
||
|
||
# +-----+------------------+
|
||
|
||
# |store| values |
|
||
|
||
# +-----+------------------+
|
||
|
||
# |1 |[1, 2, 3, 4, 5, 6]|
|
||
|
||
# |2 |[2, 3] |
|
||
|
||
# +-----+------------------+
|
||
```
|
||
|
||
## Spark SQL删除array中的指定值
|
||
|
||
1. 需求：删除dsm_time_list中代表未发生DSM报警的”0”；
|
||
|
||
2. 方法：使用array_remove函数；
|
||
|
||
```python
|
||
df_join_tmp1 = df_join_tmp1.withColumn("dsm_time_list",
|
||
|
||
flattenUdf("dsm_time_list").alias("dsm_time_list"))\
|
||
|
||
.withColumn("dsm_time_list",
|
||
|
||
F.array_remove(F.col("dsm_time_list"), "0"))
|
||
```
|
||
|
||
## 合并后的数据出现重复记录
|
||
|
||
1. 问题：在完成新增数据合并后，使用distinct函数将重复数据去除，但是不成功；
|
||
|
||
2. 原因：在取endTime和endMileage列的值时，错误使用了order命令，而max函数取的是到当前行为止之前所有数据的最大值，从而导致新增数据和原有数据有两个不同的endTime和endMileage值，因此无法使用distinct函数删除；
|
||
|
||
3. 解决方法：应按如下代码取endTime和endMileage
|
||
|
||
```python
|
||
df_union_joinTmp1=df_union.withColumn("et",F.max("endTime").over(Window.partitionBy("vin")))\
|
||
.withColumn("e_mil",F.max("endMileage").over(Window.partitionBy("vin")))
|
||
```
|
||
|
||
## 设置Spark的log等级
|
||
|
||
1. Spark默认会打印Info等级的log，导致非常多的log产生；
|
||
|
||
2. 解决方法：使用setLogLevel设置log等级；
|
||
|
||
```python
|
||
spark = SparkSession \
|
||
|
||
.builder \
|
||
|
||
.getOrCreate()
|
||
|
||
spark.sparkContext.setLogLevel("Warn")
|
||
```
|
||
|
||
## torage.DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file
|
||
|
||
1. 问题：exec在运行中报错：
|
||
|
||
|
||
```shell
|
||
21/11/09 17:35:59 ERROR storage.DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file /data5/yarn/nm/usercache/faw_vhm_admin/appcache/application_1626872989805_21612/blockmgr-ef0c9858-08bc-48b9-b9a3-27fdcfe32ef0/17/temp_shuffle_1c325963-4015-4d79-b44f-5e971de31e10
|
||
|
||
java.nio.channels.ClosedByInterruptException
|
||
|
||
at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202)
|
||
|
||
at sun.nio.ch.FileChannelImpl.truncate(FileChannelImpl.java:372)
|
||
|
||
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose$2.apply$mcV$sp(DiskBlockObjectWriter.scala:218)
|
||
|
||
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1417)
|
||
|
||
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:121)
|
||
|
||
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407)
|
||
|
||
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408)
|
||
|
||
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413)
|
||
|
||
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
|
||
|
||
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.```shell
|
||
```
|
||
|
||
2. 原因：Spark本身bug([https://issues.apache.org/jira/browse/SPARK-28340](https://issues.apache.org/jira/browse/SPARK-28340))；
|
||
|
||
3. 解决：Spark 3.0已修复该bug；
|
||
|