2024年7月 – gitweixin

运维 7月 25,2024

解决aide严重影响大数据计算时间问题

大数据离线数仓上线后，由于源头数据倍增到几十亿，发现有的耗时任务跑了几个小时也跑不出结果。明明服务器配置不错，计算内存也还可以。调大计算资源内存后也发现无济于事。

后来发现服务器有aide在运行，严重影响磁盘IO。

iotop -oP
Total DISK READ : 234.02 M/s | Total DISK WRITE : 27.09 M/s
Actual DISK READ: 238.46 M/s | Actual DISK WRITE: 11.52 M/s
PID PRIO USER DISK READ DISK WRITE SWAPIN IO> COMMAND

49757 be/4 unbound 239.94 K/s 0.00 B/s 0.00 % 99.99 % du -sk /data/dfs/dn/current/BP-1594034144-10.0.0.1-1704683739059
33597 be/4 root 7.50 M/s 0.00 B/s 0.00 % 83.99 % aide –check
54216 be/4 root 9.19 M/s 0.00 B/s 0.00 % 83.77 % aide –check
13435 be/4 root 1622.13 K/s 0.00 B/s 0.00 % 83.35 % aide –check
7996 be/4 root 3.06 M/s 0.00 B/s 0.00 % 82.80 % aide –check
25673 be/4 root 8.77 M/s 0.00 B/s 0.00 % 82.68 % aide –check
25721 be/4 root 9.50 M/s 0.00 B/s 0.00 % 81.80 % aide –check
60644 be/4 root 2.11 M/s 0.00 B/s 0.00 % 81.00 % aide –check
44128 be/4 root 10.46 M/s 0.00 B/s 0.00 % 80.85 % aide –check
3670 be/4 root 10.14 M/s 0.00 B/s 0.00 % 80.18 % aide –check
38900 be/4 root 3.06 M/s 0.00 B/s 0.00 % 79.62 % aide –check
46920 be/4 root 9.72 M/s 0.00 B/s 0.00 % 79.49 % aide –check
36099 be/4 root 10.14 M/s 0.00 B/s 0.00 % 79.35 % aide –check
32724 be/4 root 10.46 M/s 0.00 B/s 0.00 % 79.04 % aide –check
21047 be/4 root 9.50 M/s 0.00 B/s 0.00 % 78.96 % aide –check
51881 be/4 root 12.46 M/s 0.00 B/s 0.00 % 77.87 % aide –check
13147 be/4 root 10.77 M/s 0.00 B/s 0.00 % 77.56 % aide –check
36436 be/4 root 10.56 M/s 0.00 B/s 0.00 % 77.34 % aide –check

原来， AIDE 是一款入侵检测工具，它的作用是监控文件系统的完整性，防止未经授权的更改。听起来很不错，对吧？但问题就出在这里。AIDE 在工作时，需要频繁地读取和比对磁盘上的大量文件信息，这就导致了磁盘 IO 操作的大幅增加。

而大数据计算过程，如果内存不够时，需要缓存到磁盘，这时AIDE占用了大量磁盘IO，就会严重拖慢整个大数据计算的进度。

不运行AIDE后，发现耗时的离线计算运行时间缩短为之前的几分之一。

作者 east

Spark 7月 18,2024

解决pyspark的py4j.protocol.Py4JError: An error occurred while calling o84.getstate错误

在pyspark使用了udf，然后就报错下面错误：

 File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\serializers.py", line 587, in dumps
return cloudpickle.dumps(obj, 2)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 863, in dumps
cp.dump(obj)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 260, in dump
return Pickler.dump(self, obj)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 409, in dump
self.save(obj)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 736, in save_tuple
save(element)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 400, in save_function
self.save_function_tuple(obj)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 549, in save_function_tuple
save(state)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 847, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 847, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 634, in save_reduce
save(state)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 852, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 496, in save
rv = reduce(self.proto)
File "D:\ProgramData\Anaconda3\lib\site-packages\py4j\java_gateway.py", line 1257, in call
answer, self.gateway_client, self.target_id, self.name)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\sql\utils.py", line 63, in deco
return f(*a, **kw)
File "D:\ProgramData\Anaconda3\lib\site-packages\py4j\protocol.py", line 332, in get_return_value
format(target_id, ".", name, value))
py4j.protocol.Py4JError: An error occurred while calling o84.getstate. Trace:
py4j.Py4JException: Method getstate([]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
at py4j.Gateway.invoke(Gateway.java:274)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)

Traceback (most recent call last):
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\serializers.py", line 587, in dumps
return cloudpickle.dumps(obj, 2)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 863, in dumps
cp.dump(obj)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 260, in dump
return Pickler.dump(self, obj)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 409, in dump
self.save(obj)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 736, in save_tuple
save(element)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 400, in save_function
self.save_function_tuple(obj)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\cloudpickle.py", line 549, in save_function_tuple
save(state)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 847, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 847, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 634, in save_reduce
save(state)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 852, in _batch_setitems
save(v)
File "D:\ProgramData\Anaconda3\lib\pickle.py", line 496, in save
rv = reduce(self.proto)
File "D:\ProgramData\Anaconda3\lib\site-packages\py4j\java_gateway.py", line 1257, in call
answer, self.gateway_client, self.target_id, self.name)
File "D:\ProgramData\Anaconda3\lib\site-packages\pyspark\sql\utils.py", line 63, in deco
return f(*a, **kw)
File "D:\ProgramData\Anaconda3\lib\site-packages\py4j\protocol.py", line 332, in get_return_value
format(target_id, ".", name, value))
py4j.protocol.Py4JError: An error occurred while calling o84.getstate. Trace:
py4j.Py4JException: Method getstate([]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
at py4j.Gateway.invoke(Gateway.java:274)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):

刚开始以为pyspark的包有问题，后来仔细研究一下，才发现是下面的问题：

这个错误通常发生在尝试在 PySpark 中注册一个用户定义的函数（UDF）时，而该函数使用了不支持序列化的对象或方法。错误信息中的 __getstate__ 方法缺失提示 Py4J 无法正确地将 Python 对象转换为 Java 对象，这通常是因为 UDF 内部引用了 Spark DataFrame 函数或某些类实例，这些在 Py4J 上下文里是不可序列化的。

原来是在UDF用了lag 函数，这是不允许的，因为 UDF 必须是可序列化的，而 DataFrame 函数是不允许。

由于需要使用lag函数，解决方法是不用UDF来实现。

作者 east

储能 7月 15,2024

独立储能电站调峰和调频的区别

在独立储能电站中，调峰（peak shaving）和调频（frequency regulation）是两种不同的服务，它们各自在电力系统的稳定运行中扮演着重要角色。下面我将详细解释两者之间的区别：

调峰（Peak Shaving）

目的：调峰的主要目的是处理电力需求的峰值，即削峰填谷，减少电力系统在高峰时段的负荷压力。这有助于避免在高峰时段增加额外的发电容量，从而节省成本。
运作机制：在高峰用电时段，储能系统释放之前储存的能量，向电网供电，以补充发电量不足的情况；而在低谷时段，储能系统则从电网吸收多余的能量进行充电，储存起来供后续使用。
周期性：调峰通常与一天内或一周内的负荷曲线相关联，如工作日的白天用电高峰和夜晚用电低谷。
响应时间：调峰响应时间相对较长，因为它的操作通常基于预测的负荷曲线和预先规划的策略。

调频（Frequency Regulation）

目的：调频的主要目的是保持电网频率的稳定。由于电力系统中的供需必须时刻平衡，任何供需的瞬时不平衡都会导致频率波动。调频确保了这一平衡，保持电力系统的频率在一个允许的范围内，通常是50Hz或60Hz。
运作机制：储能系统在检测到电网频率偏移时，能够迅速响应，提供或吸收能量以帮助恢复频率至设定点。这种响应通常是在毫秒级或秒级完成的，比调峰更快速。
周期性：调频的周期性较短，可能在几分钟甚至几秒钟内多次发生，因为它应对的是瞬时的供需变化。
响应时间：调频要求极快的响应时间，以便及时纠正频率偏差，避免系统稳定性问题。

总结

调峰关注的是长期负荷管理，处理的是可预测的负荷高峰和低谷。
调频关注的是即时的供需平衡，处理的是不可预测的、瞬时的供需波动。

独立储能电站通过参与调峰和调频服务，不仅提高了电力系统的灵活性和可靠性，还为自己创造了额外的收入来源，因为这些服务在市场上往往是有偿的。例如，四川的“两个细则”中就提出了对独立储能进行调峰补偿，补偿标准为400元/MWh，这表明了调峰服务的市场价值。

作者 east

mysql 7月 15,2024

mysql使用 docker-compose 方式部署时区不是东八区的解决方案

mysql使用docker-compose安装后，发现时区不是东八区，导致一些服务有些出现问题。

查询时区：

SHOW VARIABLES LIKE ‘%time_zone%’;

MySQL 使用的是 UTC 时区，而不是东八区（CST）。在 Docker 容器中运行的 MySQL 实例也会使用 UTC 时区。要解决这个问题，您需要在 MySQL 配置文件中设置时区为东八区。

解决方案：

修改 my.cnf 配置文件，添加以下行以设置时区为东八区（CST）：在 [mysqld] 部分添加：

default-time_zone = '+08:00'

mysql库 /opt/mysql-master/conf/my.cnf

示例：

[mysqld]

...

default-time_zone = '+08:00'

重新启动 MySQL 容器以应用更改。

docker-compose -f /opt/mysql-master/docker-compose.yml down docker-compose -f /opt/mysql-master/docker-compose.yml up -d

现在，您的 MySQL 实例应该已经使用东八区（CST）时区了。您可以通过登录到 MySQL 并运行以下命令来验证时区设置：

SHOW VARIABLES LIKE 'time_zone';

输出应显示 time_zone 为 +08:00。

作者 east

数据仓库 7月 15,2024

大数据质量监控方法与实现

一、引言

在大数据时代，数据的质量直接关系到企业决策的准确性和业务发展的稳定性。本文旨在详细介绍大数据环境下数据质量的标准、监控方法以及相应的代码实现，确保数据的准确性、完整性、一致性和可靠性。我们将结合具体中间件和代码示例，全面阐述如何实现高效的数据质量监控。

二、数据质量标准

数据质量通常通过以下几个维度来衡量：

准确性：数据应真实反映实际情况，无错误或偏差。
完整性：数据应包含所有必需的信息，无遗漏。
一致性：同一实体在不同数据源或不同时间点的数据应保持一致。
时效性：数据应及时更新，满足业务需求。
可用性：数据应易于访问和使用，无格式或权限障碍。

三、数据质量监控方法

数据质量监控可以从多个层次进行，包括任务基线级别、任务级别与表级别、字段级别以及报表级别。

1. 任务基线级别监控

任务基线级别监控主要关注整个数据流水线（ETL任务）的运行状态和产出情况。

监控内容：
- 所有任务运行时长：与昨天运行时长对比，异常则报警。
- 结果任务产出时间：与基线规定时间对比，未按时产出则预警。

实现方式：

使用Apache Airflow等调度工具管理ETL任务，通过任务日志和执行时间监控任务运行时长和产出时间。
配置Airflow的DAG（Directed Acyclic Graph）依赖关系，确保任务按序执行。

2. 任务级别 & 表级别监控

任务级别和表级别监控关注单个任务或表的运行状态和产出数据。

监控内容：
- 任务运行时长：与昨天运行时长对比。
- 任务产出时间：与任务规定产出时间对比。
- 表产出大小：与昨日分区大小对比。

实现方式：

在ETL任务中添加日志记录功能，记录任务开始时间、结束时间和产出数据大小。
使用Shell脚本或Python脚本定期检查日志文件，对比任务运行时长、产出时间和产出大小，异常则发送邮件或消息通知。

3. 字段级别监控

字段级别监控关注具体数据字段的质量，包括指标字段和维度字段。

监控内容：
- 指标字段：均值、最大值、最小值、中位数等，与昨天、近7天、近30天的数据进行对比。
- 维度字段：维度覆盖率、维度占比、维度下指标的波动。

实现方式：

使用Apache Spark进行大规模数据处理，通过Spark SQL计算字段的统计指标。
结合Deequ等开源数据质量监控工具，编写DQC（Data Quality Center）任务，自动化监控字段质量。

代码示例：使用Deequ监控字段质量

import com.amazon.deequ.VerificationSuite
import com.amazon.deequ.checks.Check
import com.amazon.deequ.checks.CheckLevel

val verificationSuite = VerificationSuite()
  .onData(spark.table("your_table"))
  .addCheck(
    Check(CheckLevel.Error, "Mean of metric field")
      .isComplete("metric_field")
      .hasMeanEqualTo(0.5, 0.01)
  )
  .addCheck(
    Check(CheckLevel.Warning, "Dimension coverage of gender")
      .isComplete("gender")
      .hasMin("gender", 0.9)
  )

val result = verificationSuite.run()
result.show()

4. 报表级别监控

报表级别监控将监控结果可视化，便于项目组所有人直观查看数据变化。

实现方式：
- 使用Tableau、Power BI等数据可视化工具，将监控结果绘制成趋势图、仪表盘等。
- 配置定时任务，自动更新报表数据。

作者 east

Hive, Spark 7月 7,2024

Hive、Spark SQL与Impala在大数据处理中的性能对比及应用分析

一、引言

在使用CDH6.3.2处理离线报表时，Hive、Spark SQL与Impala是三种主流的大数据处理工具。下面将对这三种工具进行详细的对比分析优缺点和适用场景。

二、Hive：Apache Hive

性能分析：Hive使用MapReduce作为其数据处理后端，适用于处理大规模数据集的批量查询和分析。然而，由于MapReduce的特性，Hive在处理实时或交互式查询时性能较差。一项由Hortonworks进行的测试显示，在处理1TB数据集时，Hive的平均查询时间约为20分钟。
优点：Hive提供了类似SQL的查询语言HiveQL，使得用户可以轻松地进行数据查询和管理。此外，Hive还支持多种数据格式，包括文本文件、序列文件、Avro等。
缺点：Hive的主要缺点是其查询速度慢，不适合实时或交互式查询。此外，Hive的资源消耗也较大，需要较大的内存和磁盘空间。
适用场景：Hive适合用于处理大规模数据集的批量查询和分析，例如日志分析、用户行为分析等。

三、Spark SQL：Apache Spark SQL

性能分析：Spark SQL使用Spark作为其数据处理后端，相较于Hive，Spark SQL的查询速度更快。一项由Databricks进行的测试显示，在处理1TB数据集时，Spark SQL的平均查询时间约为2分钟，比Hive快了近10倍。
优点：Spark SQL不仅查询速度快，而且支持实时查询和交互式查询。此外，Spark SQL还支持多种数据源，包括HDFS、Hive、Parquet、JSON、JDBC等。
缺点：Spark SQL的主要缺点是其资源消耗较大，需要较大的内存和CPU资源。此外，Spark SQL的学习曲线也比Hive陡峭。
适用场景：Spark SQL适合用于处理大规模数据集的实时查询和交互式查询，例如实时数据分析、交互式数据探索等。

四、Impala：Cloudera Impala

性能分析：Impala使用MPP（大规模并行处理）架构，相较于Hive和Spark SQL，Impala的查询速度更快。一项由Cloudera进行的测试显示，在处理1TB数据集时，Impala的平均查询时间约为1分钟，比Spark SQL快了一倍。
优点：Impala不仅查询速度快，而且支持实时查询和交互式查询。此外，Impala还支持多种数据格式，包括Parquet、ORC、Avro等。
缺点：Impala的主要缺点是其资源消耗较大，需要较大的内存和CPU资源。此外，Impala的兼容性也比Hive和Spark SQL差，不支持所有的Hive表类型和函数。用Impala做ETL时，会出现显示执行成功，但时间没有把查询结果成功写入。
适用场景：Impala适合用于处理大规模数据集的实时查询和交互式查询，例如实时数据分析、交互式数据探索等。

五、结论

总的来说，Hive、Spark SQL与Impala各有优劣。Hive适用于处理大规模数据集的批量查询和分析，其查询语言易于理解，但查询速度相对较慢。Spark SQL则在查询速度上有了显著提升，同时支持实时和交互式查询，但在资源消耗上相对较高。Impala在查询速度上更胜一筹，尤其适合实时和交互式查询，但其资源消耗和兼容性是其主要缺点。

在选择这些工具时，应根据具体的应用场景和需求来决定。例如，如果是在进行大规模数据的批量查询和分析，且对查询速度的要求不高，那么Hive是一个不错的选择。如果是在进行大规模数据的实时查询和交互式查询，且对查询速度有较高的要求，那么Spark SQL或Impala可能更适合。

此外，还应考虑到系统的资源状况。如果系统资源充足，那么可以选择Spark SQL或Impala，但如果系统资源有限，那么可能需要考虑Hive或其他资源消耗较小的工具。

总的来说，Hive、Spark SQL与Impala都是优秀的大数据处理工具，它们各有优势和劣势，没有绝对的好坏之分，只有是否适合具体的应用场景和需求。

在实际应用中，我们也可以考虑将这些工具结合使用，以发挥各自的长处。例如，可以使用Hive进行大规模数据的批量查询和分析，使用Spark SQL或Impala进行实时和交互式查询，或者使用Spark SQL进行数据预处理，然后使用Hive或Impala进行数据查询和分析。

作者 east

月度归档7月 2024