2023年7月 – gitweixin

Spark 7月 28,2023

使用日期字段来实现数仓每月算一次的功能

在数仓开发中，为了实现对某个宽表每月执行一次计算的功能，由于宽表是关联日期维度表，有字段 is_end_month 可以判断是否是月末。

为了使调度任务简单，可以每天执行一次，判断当天不是月末是不执行具体计算任务，只有是月末时才执行。

// 假设你已经创建了SparkSession对象，命名为spark
import org.apache.spark.sql.functions.col

// 假设你的DataFrame名为df，包含is_end_month字段
val df = spark.read
  .format("csv")
  .option("header", "true")
  .load("path_to_your_data.csv") // 替换为你的数据路径

// 获取is_end_month字段的第一条数据
val firstRow = df.select(col("is_end_month")).first()

// 从Row对象中获取is_end_month的值，假设该字段为整数类型
val isEndMonthValue = firstRow.getAs[Int]("is_end_month")

// 判断is_end_month的值并执行相应操作
if (isEndMonthValue == 0) {
    spark.stop() // 退出Spark
} else if (isEndMonthValue == 1) {
    // 继续执行其他代码
    // ...
}
注意替换代码中的数据路径为你实际的数据文件路径，同时根据实际的数据类型对getAs方法中的类型进行调整。此代码将首先从DataFrame中选择is_end_month字段的第一行数据，然后将该字段的值提取为整数并进行判断，如果值为0，则停止Spark并退出，否则继续执行其他代码。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

Spark 7月 27,2023

spark任务运行出错：Container [pid=6978,containerID=container_1688954893175_0080_04_000001] is running 11698176B beyond the ‘PHYSICAL’ memory limit. Current usage: 1.5 GB of 1.5 GB physical memory used; 3.3 GB of 3.1 GB virtual memory used. Killing container.

运行spark任务报错，查看错误日志如下：
Failing this attempt.Diagnostics: [2023-07-25 09:57:03.987]Container [pid=6978,containerID=container_1688954893175_0080_04_000001] is running 11698176B beyond the ‘PHYSICAL’ memory limit. Current usage: 1.5 GB of 1.5 GB physical memory used; 3.3 GB of 3.1 GB virtual memory used. Killing container. Dump of the process-tree for container_1688954893175_0080_04_000001 :

根据错误日志，可以看出 Spark 任务执行过程中出现了内存问题。错误信息显示容器 container_1688954893175_0080_04_000001 正在超出 ‘PHYSICAL’ 内存限制。当前使用情况显示已使用了 1.5 GB 的物理内存（PHYSICAL memory），而该容器的物理内存限制是 1.5 GB，虚拟内存（virtual memory）使用了 3.3 GB，虚拟内存限制是 3.1 GB。由于超过了物理内存限制，Spark 正在终止该容器。

这种情况通常出现在 Spark 任务运行时，需要更多的内存资源，但配置的内存资源不足以满足任务的需求。这可能由以下原因导致：

数据量较大：处理的数据量超过了所分配的内存资源，导致内存不足。
计算复杂度高：Spark 任务涉及复杂的计算逻辑或涉及大量的数据转换操作，导致内存需求增加。
资源配置不合理：Spark 配置中分配给执行器（Executor）的内存资源设置过小。

解决方案：

增加物理内存：如果可行，可以在执行 Spark 任务的机器上增加物理内存，这样有更多的内存资源供 Spark 使用。
优化数据处理：考虑对数据处理逻辑进行优化，减少不必要的计算和数据转换，以降低内存需求。
调整 Spark 配置：在 Spark 任务提交时，通过 --conf 参数来调整 Executor 的内存分配。可以尝试增加 spark.executor.memory 参数的值来提高每个 Executor 的内存，如果有多个 Executor，可以适当增加 Executor 的数量。
使用分区技术：合理地对数据进行分区，以减少单个任务需要处理的数据量，从而减少内存压力。
检查资源使用情况：查看其他正在运行的任务以及系统的资源使用情况，确保没有其他任务占用了过多的资源，导致 Spark 任务无法获取足够的资源。
调整虚拟内存限制：如果虚拟内存限制过小，可以尝试增加虚拟内存限制，但这不是主要解决方案，因为虚拟内存只是在物理内存不足时充当备用。

根据具体情况选择合适的解决方案，并确保 Spark 任务有足够的内存资源来执行。如果问题持续存在，可能需要进一步分析任务的执行计划和资源使用情况，以找出更深层次的原因并进行针对性的优化。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

未分类 7月 27,2023

CDH重启cloudera-scm-agent报错：No socket could be created — ((‘127.0.0.1’, 9001): [Errno 98] Address already in use)

CDH集群重启 cloudera-scm-agent ，发现重启失败，查看日志如下：

[21/Jul/2023 19:49:10 +0000] 11006 HTTPServer Thread-2 _cplogging ERROR [21/Jul/2023:19:49:10] ENGINE Error in HTTP server: shutting down Traceback (most recent call last): File “/opt/cloudera/cm-agent/lib/python2.7/site-packages/cherrypy/process/servers.py”, line 225, in _start_http_thread self.httpserver.start() File “/opt/cloudera/cm-agent/lib/python2.7/site-packages/cheroot/server.py”, line 1326, in start raise socket.error(msg) error: No socket could be created — ((‘127.0.0.1’, 9001): [Errno 98] Address already in use)

错误表明 cloudera-scm-agent 在启动时无法创建HTTP服务器的套接字，原因是地址 127.0.0.1:9001 已经被占用了。这通常意味着另一个进程已经在监听相同的IP地址和端口号，导致 cloudera-scm-agent 无法绑定到该地址。

解决方案：

要解决这个问题，可以采取以下步骤：

查找占用端口的进程： 使用以下命令查找占用9001端口的进程：bash复制代码sudo netstat -tlnp | grep 9001这将显示占用9001端口的进程的相关信息，包括进程ID (PID) 和进程名。
停止占用端口的进程： 如果上一步找到了占用9001端口的进程，可以尝试停止该进程，释放端口。使用以下命令杀死该进程（请注意替换 <PID> 为实际的进程ID）：bash复制代码sudo kill <PID>或者，如果是已知的服务进程，可以使用该服务的控制命令来停止服务。
修改端口号： 如果停止占用端口的进程不可行或不适合你的情况，可以考虑修改 cloudera-scm-agent 的配置，将其绑定到一个空闲的端口上。你可以编辑配置文件中的端口号，然后重新启动 cloudera-scm-agent。
确保正确配置： 确保 cloudera-scm-agent 的配置正确，并没有指定错误的端口号。
防火墙和安全组规则： 检查服务器的防火墙设置和网络安全组规则，确保端口号没有被阻止。
检查其他应用程序： 确保没有其他应用程序意外地使用了9001端口。
重新启动服务： 尝试重新启动 cloudera-scm-agent 服务。

作者 east

运维 7月 26,2023

linux提示 netstat: command not found要怎样安装

如果你在运行 netstat 命令时出现 “netstat: command not found” 的错误提示，这说明你的系统可能没有安装 netstat 工具。在一些最新的Linux发行版中，netstat 命令已经被弃用，取而代之的是其他工具，例如 ss (socket statistics)。

如果你使用的是较新的Linux发行版，可以尝试使用 ss 命令来代替 netstat，它提供了类似的功能。以下是使用 ss 命令来查看监听端口的方法：

bash复制代码sudo ss -tlnp | grep 9001

如果你仍然需要使用传统的 netstat 命令，你可以通过安装net-tools软件包来获得它。根据你的Linux发行版，可以使用以下命令来安装net-tools：

Ubuntu / Debian：

bash复制代码sudo apt-get update
sudo apt-get install net-tools

CentOS / RHEL：

bash复制代码sudo yum install net-tools

Fedora：

bash复制代码sudo dnf install net-tools

请注意，net-tools包含了很多网络工具，其中包括 netstat 命令。安装后，你应该能够正常使用 netstat 命令了。然而，如果你的Linux发行版已经不再支持 netstat 命令，建议使用 ss 命令来查看监听端口和网络连接的信息。

作者 east

bug清单 7月 26,2023

Datart页面报错：Handler dispatch failed; nested exception is java.lang.OutOfMemoryError: GC overhead limit exceeded

错误信息 “Handler dispatch failed; nested exception is java.lang.OutOfMemoryError: GC overhead limit exceeded” 表明datart应用程序在处理请求时发生了Java虚拟机（JVM）的内存溢出错误。具体来说，JVM无法及时回收垃圾对象，导致垃圾回收的开销超过了允许的阈值，从而触发了”GC overhead limit exceeded”错误。

可能的原因：

内存不足：datart应用程序所在的JVM分配的内存不足，导致垃圾回收无法正常进行，从而引发该错误。
内存泄漏：datart应用程序可能存在内存泄漏问题，导致大量对象无法被正确回收，最终耗尽了可用内存。
处理大数据量：datart应用程序在处理大规模数据或复杂查询时，可能导致内存占用过高，超过JVM限制。

解决方案：

针对上述可能的原因，可以采取以下步骤来解决问题：

增加JVM内存限制：增加datart应用程序所在JVM的内存限制，使其能够处理更大的数据量。这可以通过修改JVM启动参数中的-Xmx和-Xms选项来实现。例如，将-Xmx选项设置为较大的值，比如 “-Xmx4g” 表示最大可用内存为4GB。
检查内存泄漏：进行内存泄漏分析，查找可能导致内存泄漏的代码，并修复问题。可以使用一些Java内存分析工具（如VisualVM、MAT等）来辅助查找内存泄漏。
优化查询：对datart应用程序中的复杂查询进行优化，尽量减少内存占用。可以通过索引优化、查询优化等方式来改善查询性能。
分页查询：如果datart应用程序处理大数据量的查询，可以考虑引入分页查询，避免一次性加载过多数据到内存中。
升级应用程序：检查datart应用程序是否有已知的内存相关问题，并考虑升级到修复了这些问题的版本。
监控和警报：设置合适的监控和警报机制，当内存使用超过阈值时及时发出警报，以便及早发现和解决问题。
分析日志：查看datart应用程序的日志，特别是错误日志，以获取更多有关错误发生时的上下文信息，有助于进一步定位问题。

作者 east

bug清单, Hadoop 7月 25,2023

Transport-level exception trying to monitor health of NameNode at xxx: java.net.SocketTimeoutException: 45000 millis timeout while waiting for the channel to be ready for read

表明CDH 6.3.2中的某个组件（可能是其他节点的Datanode或NodeManager）在尝试监视位于CDH节点上的NameNode时，发生了Socket超时异常。这意味着在连接到NameNode时花费的时间超过了45秒，导致连接失败。

可能的原因：

网络问题：有可能是网络连接不稳定或者网络延迟导致连接超时。
资源不足：CDH的NameNode可能资源不足，导致响应变慢，从而引发超时异常。
防火墙或安全设置：防火墙或其他安全设置可能限制了节点之间的通信，导致连接超时。

解决方案：

针对上述可能的原因，可以采取以下步骤逐一排查和解决问题：

检查网络连接：确保所有节点之间的网络连接稳定，并且没有阻止节点之间通信的防火墙或其他网络限制。
检查资源：确认CDH上的NameNode是否具有足够的资源（CPU、内存、磁盘空间等）来处理请求。如果资源不足，可以考虑增加资源或优化配置。
检查防火墙和安全设置：确保防火墙或其他安全设置不会阻止节点之间的通信。可以检查防火墙规则和CDH安全配置。
检查NameNode日志：查看CDH上NameNode的日志，了解是否存在其他异常或错误信息，这可能有助于进一步定位问题。
调整超时时间：可以尝试增加超时时间，从而允许更长的连接时间。但这并不是根本解决问题的方法，只是一个临时调整。
更新或升级：如果发现该问题是由于已知的CDH或Hadoop bug引起的，可以尝试升级CDH版本或应用相关的补丁和更新。
联系支持：如果上述步骤无法解决问题，可以联系CDH或Hadoop的支持团队寻求进一步的帮助和调查。

作者 east

bug清单, Hadoop 7月 25,2023

CDH节点报“Role not started due to unhealthy host”，重启不了角色

CDH集群的 Datanode 挂掉了，要重新启动报错“Role not started due to unhealthy host”。查了一下，这表示主机处于不健康状态。这个错误大概有下面的原因：

可能的原因：

主机故障：Datanode所在的主机可能存在硬件故障或者网络问题，导致主机处于不可用状态，从而Datanode无法正常启动。
资源不足：主机资源（例如CPU、内存、磁盘空间）不足，导致Datanode启动失败。
防火墙或安全设置：防火墙或其他安全设置可能会阻止Datanode与其他节点进行通信，导致启动失败。
CDH组件问题：CDH组件可能出现问题，导致Datanode无法启动。

解决方案：

针对上述可能的原因，可以采取以下步骤逐一排查和解决问题：

检查主机状态：确保Datanode所在的主机处于健康状态，没有硬件故障或网络问题。可以通过运行系统命令或者在CDH管理界面查看主机状态。
检查资源：确认主机具有足够的资源（CPU、内存、磁盘空间等）来运行Datanode。如果资源不足，可以考虑升级主机或释放资源。
检查防火墙和安全设置：确保防火墙或其他安全设置不会阻止Datanode与其他节点进行通信。可以检查防火墙规则和CDH安全设置。
检查CDH组件状态：检查CDH的其他组件是否正常运行，特别是与Datanode相关的组件（如HDFS）。如果其他组件也出现问题，可能是由于CDH整体环境的故障。
查看日志：检查Datanode日志，通常在CDH的日志目录下，查看是否有相关错误信息提供更多线索。
重启服务：尝试重启Datanode服务，以便它重新连接到集群并解决任何临时问题。
联系支持：如果上述步骤无法解决问题，可以联系CDH或Hadoop支持团队寻求帮助。

后来还发现奇怪现象，jps查到datanode、namenode进程想要kill掉，一直kill不掉。经过排查，发现是服务器多块硬盘中其中一块坏了，导致CDH一直报
“Role not started due to unhealthy host ”。

作者 east

Flink 7月 23,2023

Flink SQL 秘密：掌握变更日志事件无序的艺术

简介

Alice 是一名数据工程师，负责公司的实时数据处理。她发现Flink SQL有时会产生更新（关于键）事件。但是，在 Flink 的早期版本中，这些事件无法直接写入 Kafka，因为 Kafka 本质上是一个仅追加的消息系统。幸运的是，Flink 社区在更高版本中发布了 Connectorupsert-kafka，支持写入更新事件。后来她发现Flink SQL作业readupsert-kafkaevents进行join操作时偶尔会出现错误。这让她对Flink SQL的可靠性产生了怀疑。她向社区报告了该问题，确认是变更日志事件乱序问题，随后在新版本中得到了解决。终于，她又可以继续愉快地使用 Flink SQL 了。从 Alice 使用 Flink SQL 的经历中我们可以了解到，实时数据处理并不总是顺利和直接的。为了让 Flink SQL 更容易理解，本文试图解开 Flink SQL 中变更日志事件乱序问题的谜团。我们首先介绍 Flink SQL 中的changelog，然后演示changelog事件乱序问题及其解决方案。最后，我们将针对这个问题提出最佳实践，以帮助您更好地理解和使用 Flink SQL 进行实时数据处理。

Flink SQL 中的 ChangelogChangelog 并不是 Flink SQL 发明的新概念。在关系数据库世界中，MySQL使用众所周知的binlog（二进制日志）来记录数据库中的所有修改操作，包括INSERT、UPDATE和DELETE操作。同样，Flink SQL 中的changelog 也用于记录这些数据变化，以实现增量数据处理。在 MySQL 中，binlog 可以用于数据的备份恢复、同步和复制。通过读取并解析binlog中的操作记录，可以实现增量数据的同步和复制。变更数据捕获（CDC）是一种常用的数据同步技术，它监视数据库中的数据变化并将这些变化转换为事件流进行实时处理。 CDC工具可用于将关系数据库中的数据变化实时传输到其他系统或数据仓库，以支持实时分析和报告。常见的CDC工具包括Debezium和Maxwell。通过 FLINK-15331 添加的 Flink CDC 支持，允许与外部系统的 CDC 数据实时集成，并通过 Flink 实现实时数据同步和分析。在 Flink SQL 中生成和处理 Changelog 事件。虽然前面提到的 binlog 和 CDC 是与 Flink 集成的外部 Changelog 数据源，但 Flink SQL 内部也会生成 Changelog 数据。为了区分事件是否是更新事件，我们将仅包含 INSERT 类型事件的变更日志称为附加流，而另外包含其他类型（例如 UPDATE）事件的变更日志称为更新流。

Flink 中的一些操作（例如组聚合、去重）会产生更新事件。生成更新事件的运算符通常会维护状态，我们通常将它们称为有状态运算符。值得注意的是，并非所有有状态运算符都支持将更新流作为输入进行处理。例如，窗口聚合和间隔连接目前还不支持更新流作为输入。

熟悉Debezium数据格式（或数据库binlog解析）的用户可能想知道为什么Flink不使用复合UPDATE事件类型（就像数据库所做的那样），既包含UPDATE_BEFORE（UB）又包含UPDATE_AFTER（UA）并且更紧凑。事实上，我们在设计和实现 Flink 的回缩机制时就仔细评估了这个选项。复合 UPDATE 事件在某些场景下确实更加紧凑，可以解决特定问题（例如 FLINK-9528），但我们选择不使用它的原因主要是以下两个方面：拆分事件无论事件类型如何，其事件结构都是相同的（仅 RowKind 不同），这使得序列化更简单。如果使用复合 UPDATE 事件，则事件要么是异构的，要么 INSERT/DELETE 也被建模为 UPDATE 事件（例如，INSERT 事件仅具有 UA，DELETE 事件仅具有 UB）。在分布式环境中，经常涉及数据混洗操作（例如，联接、聚合）。即使使用复合 UPDATE 事件，在某些场景中进行混洗时，它们仍然必须拆分为单独的 DELETE 和 INSERT 事件。

作者 east

doris 7月 18,2023

java代码判断doris的某一天的分区是否存在

需要使用doris的SHOW PARTITIONS命令来查看表的分区信息。这个命令会显示表的所有分区信息，包括分区名、范围、副本数等。您可以根据您设定的动态分区规则判断是否有缺失或多余的分区 ¹。

要通过java代码执行doris的SHOW PARTITIONS命令，您可以使用以下步骤：

导入doris的jdbc驱动包，例如doris-jdbc-0.15.0.jar²。
创建一个数据库连接对象，指定doris的url、用户名和密码。
创建一个语句对象，执行SHOW PARTITIONS FROM tbl1命令，其中tbl1是您要查询的表名。
获取一个结果集对象，遍历结果集，判断某一天的分区是否存在。

以下是一个可能的java代码示例，仅供参考：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class DorisPartitionTest {

    public static void main(String[] args) throws Exception {
        // Load the driver
        Class.forName("org.apache.doris.jdbc.Driver");
        // Create a connection object
        Connection conn = DriverManager.getConnection("jdbc:doris://127.0.0.1:9030?database=test_db", "root", "");
        // Create a statement object
        Statement stmt = conn.createStatement();
        // Execute the command
        ResultSet rs = stmt.executeQuery("SHOW PARTITIONS FROM tbl1");
        // Define a variable to store the partition name of the day to be queried
        String partitionName = "p20210717";
        // Define a variable to store the result of whether the partition exists
        boolean partitionExists = false;
        // Traverse the result set
        while (rs.next()) {
            // Get the partition name column
            String name = rs.getString("PartitionName");
            // Compare with the partition name to be queried
            if (name.equals(partitionName)) {
                // If equal, set the result to true and break the loop
                partitionExists = true;
                break;
            }
        }
        // Close the resources
        rs.close();
        stmt.close();
        conn.close();
        // Print the result
        System.out.println("The partition " + partitionName + " exists: " + partitionExists);
    }
}

作者 east

Spark 7月 17,2023

spark dataframe如何提取某一行的字段的值

有几种方法可以从Spark dataframe中提取某一行的字段的值，使用scala语言实现。我会给你一些例子和链接，你可以参考一下。

一种方法是使用**row.getString(index)**方法，其中index是字段在行中的位置。例如，如果你想要获取第一行的第一个字段的值，你可以这样写：

val row = df.first() // 获取第一行
val value = row.getString(0) // 获取第一个字段的值

这个方法需要你知道字段的索引和类型。¹

另一种方法是使用**row.getAsT**方法，其中T是字段的类型，columnName是字段的名称。例如，如果你想要获取第一行的name字段的值，你可以这样写：

val row = df.first() // 获取第一行
val value = row.getAs[String]("name") // 获取name字段的值

这个方法不需要你知道字段的索引，但需要你知道字段的类型和名称。²

还有一种方法是使用**df.select(columnName).collect()**方法，其中columnName是字段的名称。这个方法会返回一个包含所有行的字段值的数组。例如，如果你想要获取所有行的name字段的值，你可以这样写：

val values = df.select("name").collect() // 获取所有行的name字段的值

复制

这个方法不需要你知道字段的索引和类型，但需要你知道字段的名称。

作者 east

月度归档7月 2023