gitweixin
  • 首页
  • 小程序代码
    • 资讯读书
    • 工具类
    • O2O
    • 地图定位
    • 社交
    • 行业软件
    • 电商类
    • 互联网类
    • 企业类
    • UI控件
  • 大数据开发
    • Hadoop
    • Spark
    • Hbase
    • Elasticsearch
    • Kafka
    • Flink
    • 数据仓库
    • 数据挖掘
    • flume
    • Kafka
    • Hive
    • shardingsphere
    • solr
  • 开发博客
    • Android
    • php
    • python
    • 运维
    • 技术架构
    • 数据库
  • 程序员网赚
  • bug清单
  • 量化投资
  • 在线查询工具
    • 去行号
    • 在线时间戳转换工具
    • 免费图片批量修改尺寸在线工具
    • SVG转JPG在线工具

年度归档2023

精品微信小程序开发门户,代码全部亲测可用

  • 首页   /  
  • 2023
  • ( 页面8 )
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命令来查看表的分区信息。这个命令会显示表的所有分区信息,包括分区名、范围、副本数等。您可以根据您设定的动态分区规则判断是否有缺失或多余的分区1。

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

  • 导入doris的jdbc驱动包,例如doris-jdbc-0.15.0.jar2。
  • 创建一个数据库连接对象,指定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) // 获取第一个字段的值

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

  • 另一种方法是使用**row.getAsT**方法,其中T是字段的类型,columnName是字段的名称。例如,如果你想要获取第一行的name字段的值,你可以这样写:
val row = df.first() // 获取第一行
val value = row.getAs[String]("name") // 获取name字段的值

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

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

复制

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

作者 east
Spark 6月 19,2023

spark dataframe left join另一个dataframe 空值异常的问题

当在Spark开发时,某个字段有空值会出现不少意想不到的情况。Spark SQL中对两个Dataframe使用join时,当作为连接的字段的值含有null值。由于null表示的含义是未知,既不知道有没有,在SQL中null值与任何其他值的比较(即使是null)永远不会为真。故在进行连接操作时null == null不为True,所以结果中不会出现该条记录,即左侧表格的这条记录对应右侧的值均为null。

解决方法一:

如果两个DataFrame进行left join时,多个字段的值有空值,那么结果就会为空。为了解决这个问题,我们可以先对这两个DataFrame进行处理,在处理的过程中将空值替换成一个特殊值,例如:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{Column, DataFrame}

// 左边的DataFrame为leftDF,右边的DataFrame为rightDF
// 给定leftDF和rightDF共同进行left join的字段列
val joinColumns: Seq[String] = Seq("col1", "col2", "col3")

// 定义替换的特殊值
val specialValue: String = "__NULL__"

// 对leftDF和rightDF的joinColumns列进行空值替换
val leftJoinDF: DataFrame = replaceNullsWithSpecialValue(leftDF, joinColumns, specialValue)
val rightJoinDF: DataFrame = replaceNullsWithSpecialValue(rightDF, joinColumns, specialValue)

// 对leftJoinDF和rightJoinDF进行join操作
val joinedDF: DataFrame = leftJoinDF.join(rightJoinDF, joinColumns, "left")

// 定义空值替换函数
def replaceNullsWithSpecialValue(df: DataFrame, columns: Seq[String], replacement: String): DataFrame = {
  val columnsToReplace: Seq[Column] = columns.map(col(_))
  val columnsToKeep = df.columns.filterNot(columns.contains(_)).map(col)
  df.select((columnsToReplace ++ columnsToKeep):_*).na.fill(replacement, columnsToReplace)
}

在这里,我们使用na.fill()函数将DataFrame中的空值替换为特殊值。在处理完之后,我们就可以对两个DataFrame进行left join操作了。

解决方法二:

一种可能的解决方案是使用NULL safe equality operator(<=>),它可以在join条件中处理NULL值,使得NULL值与NULL值相等。例如,如果你有两个dataframe,df1和df2,你想要根据多个字段进行left join,你可以写成:

import org.apache.spark.sql.functions._
val joinedDF = df1.join(df2, df1("col1") <=> df2("col1") && df1("col2") <=> df2("col2"), "left")

这样,即使col1或col2中有NULL值,也不会影响join的结果。


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


作者 east
doris, Spark 6月 16,2023

Spark写数据到Doris报错node and exceeded the max retry times

用spark dataframe向doris写数据时,报下面错误:

Failed to load data on BE: http://192.168.50.10:18040/api/mydb/dwd_virtual_table/_stream_load? node and exceeded the max retry times.

发现表没写入成功。刚开始很困惑,后来发现是 dataFrame中的字段和目标表不一致 。

这种提示很不友好,有没有更好方式提示,方法是有的,可以用jdbc写入,发现错误时可以看到具体的提示。代码参考如下:

def writeByJDBC(dataframe: DataFrame, dorisTable: String): Unit = {
    dataframe.write.format("jdbc")
      .mode(SaveMode.Append)
      .option("driver", "com.mysql.jdbc.Driver")
      .option("url", "jdbc:mysql://" + DORIS_HOST + ":9030/" +DATABASE_NAME + "?rewriteBatchedStatements=false")
      .option("batchsize", "" + WRITE_BATCH_SIZE)
      .option("user", DORIS_USER)
      .option("password", DORIS_PASSWORD)
      .option("isolationLevel", "NONE")
      //  .option("doris.write.fields","case_id,defendant_name,finance_name,mediation_name,mediator_name,dt")
      .option("dbtable", dorisTable)
      .save()
  }

不过这种方式还是没有Spark Doris Connector的方式写效率高,可以用上面jdbc方式调试,没问题再切换 Spark Doris Connector 方式:

def writeByDoris(dataframe: DataFrame, dorisTable: String): Unit = {
dataframe.write.format(“doris”)
.option(“doris.table.identifier”, dorisTable)
.option(“doris.fenodes”, DORIS_HOST + “:” + DORIS_FE_HTTP_PORT)
.option(“user”, DORIS_USER)
.option(“password”, DORIS_PASSWORD)
.option(“sink.batch.size”, WRITE_BATCH_SIZE)
.option(“sink.max-retries”, 3)
.option(“doris.request.retries”, 6)
.option(“doris.request.retries”, 100)
.option(“doris.request.connect.timeout.ms”, 60000)
.save()
}

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

作者 east
Spark 6月 16,2023

spark sql多个指标合并为1条sql

在 Spark SQL 指标开发中,使用一条 SQL 语句来计算多个指标是常见的做法。以下是几种常用的方法:

  1. 使用子查询:将多个指标的计算逻辑放在子查询中,再在外层 SELECT 语句中进行聚合。这种方式可以避免代码冗余,但可能会降低 SQL 语句的可读性。
  2. 使用 Window 函数:使用 Window 函数可以在一条 SQL 语句中计算多个指标,并且可以避免对同一个数据集进行多次扫描。具体操作是使用 over() 函数配合 sum、avg、max、min等聚合函数。这种方式相比子查询可以提升 SQL 的效率,但需要熟练掌握 Window 函数的使用方法。
  3. 使用 UDAF:如果需要计算的指标比较复杂,可以考虑开发自定义聚合函数(UDAF),这样可以将计算逻辑封装到统一的函数中,提高代码的可复用性和可维护性。

例子:

  1. 使用子查询进行多指标计算

假设我们有一个订单表 orders,包含以下字段:order_id, customer_id, order_date, amount。我们需要计算每个客户在 2022 年的订单数量和总订单金额。可以使用如下 SQL 语句实现:

SELECT 
  customer_id, 
  (SELECT COUNT(*) 
   FROM orders 
   WHERE customer_id = o.customer_id AND YEAR(order_date) = 2022) AS order_count, 
  (SELECT SUM(amount) 
   FROM orders 
   WHERE customer_id = o.customer_id AND YEAR(order_date) = 2022) AS total_amount 
FROM orders o 
GROUP BY customer_id;

这个 SQL 语句使用了两个子查询来计算每个客户在 2022 年的订单数量和总订单金额。子查询返回的结果会作为 select 语句中的一个列,因此可以使用 group by 对客户进行分组。

  1. 使用 Window 函数进行多指标计算

使用 Window 函数可以更方便地对查询结果进行分析。假设我们有一个销售表 sales,包含以下字段:sale_id, customer_id, product_id, sale_date, sale_amount。我们需要计算以下指标:每个客户的总销售额、每个客户的最大销售额、每个客户的销售额排名。可以使用如下 SQL 语句实现:

SELECT 
  customer_id, 
  SUM(sale_amount) OVER (PARTITION BY customer_id) AS total_sale, 
  MAX(sale_amount) OVER (PARTITION BY customer_id) AS max_sale, 
  RANK() OVER (PARTITION BY customer_id ORDER BY sale_amount DESC) AS sale_rank 
FROM sales;

这个 SQL 语句使用了三个 Window 函数,分别计算每个客户的总销售额、最大销售额和销售额排名。这里我们将结果按客户分组,然后使用了 Partition By 子句指定了客户维度。

  1. 使用自定义聚合函数进行多指标计算

自定义聚合函数(UDAF)可以针对特定的需求编写自己的聚合逻辑。以计算最小值、最大值和平均值为例,我们可以实现一个自定义平均值 UDAF:

import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
 
class CustomAvg extends UserDefinedAggregateFunction {
 
  // 输入参数的数据类型
  def inputSchema: StructType = new StructType().add("inputColumn", DoubleType)
 
  // 中间结果的数据类型
  def bufferSchema: StructType = new StructType().add("sum", DoubleType).add("count", LongType)
 
  // 输出结果的数据类型
  def dataType: DataType = DoubleType
 
  // 是否固定输入参数类型
  def deterministic: Boolean = true
 
  // 初始化中间结果缓存
  def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0.0 // sum
    buffer(1) = 0L // count
  }
 
  // 更新中间结果缓存
  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (!input.isNullAt(0)) {
      buffer(0) = buffer.getDouble(0) + input.getDouble(0)
      buffer(1) = buffer.getLong(1) + 1
    }
  }
 
  // 合并中间结果缓存
  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }
 
  // 计算最终结果
  def evaluate(buffer: Row): Any = {
    buffer.getDouble(0) / buffer.getLong(1)
  }
}

这里我们实现了一个 CustomAvg UDAF,可以通过输入若干个 Double 类型的值,计算它们的平均值并返回。我们可以在 SparkSQL 中使用这个 UDAF 计算多个指标的平均值:

SELECT 
  CustomAvg(column1) AS avg1, 
  CustomAvg(column2) AS avg2, 
  CustomAvg(column3) AS avg3 
FROM table;

这个 SQL 语句使用了 CustomAvg UDAF 计算了三个列的平均值,可以根据实际需求进行扩展。

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

作者 east
doris 6月 13,2023

Spark 写入数据到doris异常

用spark写入到doris,由于服务器配置不高, sink.batch.size 设置过大时,超出服务器内存限制,doris be端直接闪崩。如果 sink.batch.size 设置过小,则出现写入次数太频繁无法写入。 调整 write 的
sink.batch.size 参数大小,逐步增加 batchsize 大小,以达到更好的写入性能与内存占用的平衡。例如,可以逐渐将 batchsize 参数的值从 200 调整到 500,1000,2000,以找到最好的性能与内存占用平衡点。 刚开始设置10000时服务器be端闪崩,后来设置200时又写到中途报错,最后找到2000这个平衡点。

def writeDoris(dataframe: DataFrame, dorisTable: String): Unit = {
dataframe.write.format(“doris”)
.option(“doris.table.identifier”, dorisTable)
.option(“doris.fenodes”, DORIS_HOST + “:” + DORIS_FE_HTTP_PORT)
.option(“user”, DORIS_USER)
.option(“password”, DORIS_PASSWORD)
.option(“sink.batch.size”, 2000)
.save()
}

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

作者 east
doris 6月 8,2023

spark运行写入doris的任务报错park.DorisStreamLoad: Doris BE’s response cannot map to schema

在CDH的集群上跑spark任务,运行报错:
ERROR spark.DorisStreamLoad: Doris BE’s response cannot map to schema. res: “Access denied for user ‘default_cluster:hdfs@10.0.10.101’ (using password: YES)”
org.apache.doris.shaded.com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of org.apache.doris.spark.rest.models.BackendV2 (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value (‘Access denied for user ‘default_cluster:hdfs@10.0.10.101′ (using password: YES)’)
at [Source: (String)””Access denied for user ‘default_cluster:hdfs@10.0.10.101’

查了一下,连接doris的用户名和密码没错。后来发现,spark-submit的参数–conf spark.user=hdfs 影响了,把这个参数去掉就正常了。

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

作者 east
doris 6月 1,2023

spark jdbc模式写入异常

用spark jdbc模式写入doris报下面错误:

java.sql.BatchUpdateException: Insert has filtered data in strict mode, tracking_url=http://192.168.1.40:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
	at com.mysql.cj.util.Util.getInstance(Util.java:167)
	at com.mysql.cj.util.Util.getInstance(Util.java:174)
	at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:755)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:426)
	at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:800)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.savePartition(JdbcUtils.scala:667)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:834)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:834)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:935)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:935)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2121)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2121)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	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:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLException: Insert has filtered data in strict mode, tracking_url=http://10.0.80.54:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129)
	at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:953)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1098)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1046)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeLargeUpdate(ClientPreparedStatement.java:1371)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:746)
	... 17 more

点开上面的=http://192.168.1.40:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2,发现报下面的错误:

Reason: no partition for this tuple. tuple=

查看要写入doris对应的表,这个表是动态分区的,发现这个表没有插入数据的分区。后来匹配对应的doris表动态分区后,插入数据果然正常了。

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

作者 east
python 5月 28,2023

解决cannot import name ‘EVENT_TYPE_OPENED’ from ‘watchdog.events’ (/root/anaconda3/lib/python3.9/site-packages/watchdog/events.py)

这个错误表明在尝试从watchdog.events模块导入EVENT_TYPE_OPENED时发生了问题。在搜索结果中,我没有找到与你的问题直接相关的解决方案。但是,我们可以尝试一些方法来解决这个问题。

首先,检查你的代码中是否正确导入了watchdog.events模块。确保你的代码中有以下导入语句:

from watchdog.events import EVENT_TYPE_OPENED

如果你的代码中已经有了这个导入语句,那么问题可能是watchdog库的版本。尝试更新watchdog库到最新版本。你可以使用以下命令来更新:

pip install --upgrade watchdog

如果更新watchdog库之后问题仍然存在,你可以尝试查看watchdog.events模块的源代码,看看EVENT_TYPE_OPENED是否存在。你可以在你的Python环境的site-packages目录下找到watchdog库。在你的问题描述中,watchdog.events模块的路径是/root/anaconda3/lib/python3.9/site-packages/watchdog/events.py。

如果在watchdog.events模块中找不到EVENT_TYPE_OPENED,那么你可能需要查找使用了EVENT_TYPE_OPENED的代码,并尝试替换为watchdog.events模块中存在的其他事件类型。你可以参考watchdog库的文档来了解更多关于事件类型的信息。

如果在尝试上述方法后问题仍然存在,请提供更多关于你的代码和环境的信息,以便我们能够更好地帮助你解决问题

作者 east
doris 5月 26,2023

datax写入doris报错:Writing records to Doris failed

用datax同步数据到doris,报下面错误:

java.lang.RuntimeException: Writing records to Doris failed.
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.close(DorisWriterManager.java:113)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriter$Task.post(DorisWriter.java:150)
 at com.alibaba.datax.core.taskgroup.runner.WriterRunner.run(WriterRunner.java:65)
 at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Writing records to Doris failed.
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.checkFlushException(DorisWriterManager.java:189)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.waitAsyncFlushingDone(DorisWriterManager.java:150)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.flush(DorisWriterManager.java:98)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.close(DorisWriterManager.java:111)
 ... 3 more
Caused by: java.io.IOException: java.io.IOException: Failed to flush data to Doris.
{"TxnId":-1,"Label":"datax_doris_writer_fed9613c-8d95-4284-a9a2-949985cf3f8d","TwoPhaseCommit":"false","Status":"Fail","Message":"errCode = 7, detailMessage = unknown table, tableName=mediation","NumberTotalRows":0,"NumberLoadedRows":0,"NumberFilteredRows":0,"NumberUnselectedRows":0,"LoadBytes":0,"LoadTimeMs":0,"BeginTxnTimeMs":0,"StreamLoadPutTimeMs":0,"ReadDataTimeMs":0,"WriteDataTimeMs":0,"CommitAndPublishTimeMs":0}
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.asyncFlush(DorisWriterManager.java:170)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.access$000(DorisWriterManager.java:19)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager$1.run(DorisWriterManager.java:134)
 ... 1 more
Caused by: java.io.IOException: Failed to flush data to Doris.

在这种情况下,错误是由于未知表引起的。这可能是由许多因素引起的,例如:

表不存在。
表不可访问。
表配置不正确。
要修复错误,您需要确定错误的原因并采取措施进行纠正。如果表不存在,则需要创建表。如果无法访问表,则需要授予用户对表的写入权限。如果表配置不正确,则需要正确配置它。

一旦您纠正了错误的原因,就应该能够无问题地将数据写入Doris中。

以下是一些解决此错误的附加提示:

检查Doris日志以获取更多有关错误的信息。
尝试使用其他工具(例如Doris CLI或Doris API)将数据写入Doris。
如果仍然遇到问题,您可以联系Doris支持团队寻求帮助。

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

作者 east
云计算 5月 25,2023

IT 项目管理基础:衡量云的成功

最重要的 IT 项目管理基础之一是衡量成功。对于基于云的 IT 项目,这尤其复杂且重要。

当您在云中运营时,很容易超支、扩展速度过快或过慢、失去可见性或犯一系列其他错误。看一个 IT 云项目是成功还是失败是至关重要的。这种可见性使企业和项目管理公司能够确定他们做对了什么,哪里做错了,以及下次可以吸取什么教训。

公司如何通过定义然后取得成功来确保他们保持 IT 项目管理基础一致?

IT 项目是任何涉及与 IT 基础设施、云计算、本地服务、信息系统等相关的投资、实施和改进的业务。它代表了大量工作,可以包括 IT 的许多领域,例如 Web 开发、软件开发、数据库管理和网络配置等等。

IT 项目经理必须了解 IT 项目会影响多个部门和垂直领域。基于云的项目也是如此。 IT 云项目跨越不同的部门和团队。其中许多可能不会经常互动。

大型企业一旦开始涉及云的 IT 项目,许多部门需要循环:安全、合规、销售等。独立的部门需要联合起来以实现整体 IT 目标。因此,一个成功的 IT 项目可以推动整体增长;失败的一个可能对整个组织有害。

IT 投资,尤其是在云计算方面的投资,已大幅增加,到 2021 年底,支出将达到近 4 万亿美元。有如此多的资金和资源专门用于 IT 云项目,了解投资回报率至关重要。这就是衡量项目成功与否的主要 IT 项目管理基础之一。

此外,随着数据收集和分析变得越来越普遍,衡量云 IT 项目的成功为所有相关方提供了一个更大、更清晰的画面,包括业务中的工作、偏离目标的内容、可以改进的内容等等。

通过从以往的成功和失败中吸取教训并将其整合到未来的项目中,您和您的团队能够减少错误,做出更准确的项目预算分配,有效地管理您的费用,实现您的项目目标,并交付预期的结果.整个组织以各种重要方式成为更精明的云用户。

由于 IT 项目的成功是 IT 项目管理基础的关键,因此了解如何衡量成功非常重要。在评估 IT 云项目是否成功时,这些 IT 项目管理指标是关键:

1. 时间表:每个 IT 项目都有一个时间表或时间框架。项目经理和团队的目标是在该时间范围内交付和移交项目,并证明他们能够准确估计完成给定项目所需的时间。

2. 生产力:这一特定指标表明资源管理在 IT 项目中的重要性。项目经理必须评估其团队对资产的整体利用率。团队是否最大化了他们的资源?他们在项目期间是否有效?总工作量是否符合预算工作量?对这些问题的积极回答意味着对底线的积极影响。未能最大限度地利用资源会导致巨大的经济损失。每年,公司都会因为闲置和未充分利用的计算资源而损失 88 亿美元。

3. 预算:许多高管将预算和成本管理列为项目的最高优先事项,因此评估项目在财务上的表现至关重要。简而言之,您是否设法将项目费用控制在预算范围内,还是超支了?对于许多企业来说,预算是一个棘手的领域。我们的 2021 年大数据云技术报告发现,对于 40% 的企业而言,云支出预计会超出预算 40% 或更多。

4. 质量和满意度:质量控制是一种关注客户及其满意度的 IT 项目管理指标。整个项目每个阶段的工作水平都应该很好,从而产生符合甚至超过标准和期望的高质量交付物。

5. 投资回报率 (ROI):ROI 是另一个关键的 IT 项目管理基本指标。通常,项目的最终目标是通过增加利润为组织做出贡献。管理人员使用 ROI 来确定项目是否会产生积极的回报,并根据项目的财务损益与项目总成本来确定项目是否具有业务价值。

每个项目通常都有不同的目标。适用于一个企业或一个业务部门的 IT 项目管理指标可能不适用于另一个企业。因此,根据您企业的独特需求定制指标和报告非常重要。

您现在能够更快地运行 Spark 作业了吗?或者在特定时间内有更多的 Spark 作业?停机时间是否低于某个阈值?

这些可能是适合您的问题,或者完全不同的指标可能是合适的。如果 IT 项目管理指标根据您企业的独特需求精心定制,则量化和衡量项目成功要容易得多。

自云计算出现以来,IT 云项目激增。在提高业务效率、提高数据安全性和减少运营支出的承诺的推动下,许多企业被鼓励将其流程和应用程序迁移到基于云的环境。

87% 的企业现在实施混合云战略,而 93% 目前实施多云方法。令人惊讶的是,大多数企业迁移到云端是为了“更好地优化他们现有的云资产”。

大型 IT 云项目代表着时间、资源、金钱和努力的投资——所有这些都具有隐含的可衡量的投资回报率。使用项目管理系统来跟踪和衡量 IT 云项目的成功势在必行。然而,完全利用本地工具来管理项目是违反直觉的,因为这种方法变得越来越过时,并进一步增加了 IT 项目的预算和支出。

选择基于云的项目管理解决方案有望取得更大的成功。除了明显的成本效益外,在云中管理项目还为项目经理和团队提供了在本地应用程序中无法获得的灵活性和可扩展性。基于云的项目管理解决方案使用户能够根据需求变化扩展团队或添加新功能。

项目管理团队现在是分散的,不再局限于办公室。拥有云托管应用程序可确保他们能够随时随地访问它,这样他们就可以与其他成员同步工作。

项目是有时间限制的努力。一旦 IT 项目到达其生命周期的终点(或未能到达终点),只需考虑三个步骤:终止、延期或过渡。

在项目的所有权和责任正式移交(或部门)之前,项目经理必须进行项目收尾。收尾是另一个重要的 IT 项目管理基础。

项目管理的最后阶段使项目团队能够重新审视项目,确保满足所有项目,并为未来的 IT 计划收集和存储尽可能多的信息。

可靠的项目收尾清单包括以下步骤:

利益相关者可能会发表评论。接受他们并向他们学习。他们的评论是 IT 项目管理基础,因为这些将帮助您在未来交付更好的项目。

每个 IT 项目都需要开发人员提供持续的 IT 支持,尤其是在过渡、入职、培训、错误、修复和更新方面。 IT 供应商可以根据他们的合同协议提供免费的基本 IT 支持或高级支持服务。

交付 IT 云项目后,您和您的团队就需要确保从云投资中获得最大价值。云托管的基础架构、流程和应用程序无疑可以加速任务、提高效率并简化产品周期等。

然而,80% 的企业最终花费超过了他们最初的云预算。为什么?他们未能优化其云基础架构性能和成本。 IT 云项目旨在减少您的运营开支,而不是增加运营开支。但您需要深入了解您的云 IT 项目,看看问题出在哪里。

作者 east

上一 1 … 7 8 9 … 19 下一个

关注公众号“大模型全栈程序员”回复“小程序”获取1000个小程序打包源码。回复”chatgpt”获取免注册可用chatgpt。回复“大数据”获取多本大数据电子书

标签

AIGC AI创作 bert chatgpt github GPT-3 gpt3 GTP-3 hive mysql O2O tensorflow UI控件 不含后台 交流 共享经济 出行 图像 地图定位 外卖 多媒体 娱乐 小程序 布局 带后台完整项目 开源项目 搜索 支付 效率 教育 日历 机器学习 深度学习 物流 用户系统 电商 画图 画布(canvas) 社交 签到 联网 读书 资讯 阅读 预订

官方QQ群

小程序开发群:74052405

大数据开发群: 952493060

近期文章

  • 如何在Chrome中设置启动时自动打开多个默认网页
  • spark内存溢出怎样区分是软件还是代码原因
  • MQTT完全解析和实践
  • 解决运行Selenium报错:self.driver = webdriver.Chrome(service=service) TypeError: __init__() got an unexpected keyword argument ‘service’
  • python 3.6使用mysql-connector-python报错:SyntaxError: future feature annotations is not defined
  • 详解Python当中的pip常用命令
  • AUTOSAR如何在多个供应商交付的配置中避免ARXML不兼容?
  • C++thread pool(线程池)设计应关注哪些扩展性问题?
  • 各类MCAL(Microcontroller Abstraction Layer)如何与AUTOSAR工具链解耦?
  • 如何设计AUTOSAR中的“域控制器”以支持未来扩展?

文章归档

  • 2025年7月
  • 2025年6月
  • 2025年5月
  • 2025年4月
  • 2025年3月
  • 2025年2月
  • 2025年1月
  • 2024年12月
  • 2024年11月
  • 2024年10月
  • 2024年9月
  • 2024年8月
  • 2024年7月
  • 2024年6月
  • 2024年5月
  • 2024年4月
  • 2024年3月
  • 2023年11月
  • 2023年10月
  • 2023年9月
  • 2023年8月
  • 2023年7月
  • 2023年6月
  • 2023年5月
  • 2023年4月
  • 2023年3月
  • 2023年1月
  • 2022年11月
  • 2022年10月
  • 2022年9月
  • 2022年8月
  • 2022年7月
  • 2022年6月
  • 2022年5月
  • 2022年4月
  • 2022年3月
  • 2022年2月
  • 2022年1月
  • 2021年12月
  • 2021年11月
  • 2021年9月
  • 2021年8月
  • 2021年7月
  • 2021年6月
  • 2021年5月
  • 2021年4月
  • 2021年3月
  • 2021年2月
  • 2021年1月
  • 2020年12月
  • 2020年11月
  • 2020年10月
  • 2020年9月
  • 2020年8月
  • 2020年7月
  • 2020年6月
  • 2020年5月
  • 2020年4月
  • 2020年3月
  • 2020年2月
  • 2020年1月
  • 2019年7月
  • 2019年6月
  • 2019年5月
  • 2019年4月
  • 2019年3月
  • 2019年2月
  • 2019年1月
  • 2018年12月
  • 2018年7月
  • 2018年6月

分类目录

  • Android (73)
  • bug清单 (79)
  • C++ (34)
  • Fuchsia (15)
  • php (4)
  • python (45)
  • sklearn (1)
  • 云计算 (20)
  • 人工智能 (61)
    • chatgpt (21)
      • 提示词 (6)
    • Keras (1)
    • Tensorflow (3)
    • 大模型 (1)
    • 智能体 (4)
    • 深度学习 (14)
  • 储能 (44)
  • 前端 (5)
  • 大数据开发 (491)
    • CDH (6)
    • datax (4)
    • doris (31)
    • Elasticsearch (15)
    • Flink (78)
    • flume (7)
    • Hadoop (19)
    • Hbase (23)
    • Hive (41)
    • Impala (2)
    • Java (71)
    • Kafka (10)
    • neo4j (5)
    • shardingsphere (6)
    • solr (5)
    • Spark (100)
    • spring (11)
    • 数据仓库 (9)
    • 数据挖掘 (7)
    • 海豚调度器 (10)
    • 运维 (34)
      • Docker (3)
  • 小游戏代码 (1)
  • 小程序代码 (139)
    • O2O (16)
    • UI控件 (5)
    • 互联网类 (23)
    • 企业类 (6)
    • 地图定位 (9)
    • 多媒体 (6)
    • 工具类 (25)
    • 电商类 (22)
    • 社交 (7)
    • 行业软件 (7)
    • 资讯读书 (11)
  • 嵌入式 (71)
    • autosar (63)
    • RTOS (1)
    • 总线 (1)
  • 开发博客 (16)
    • Harmony (9)
  • 技术架构 (6)
  • 数据库 (32)
    • mongodb (1)
    • mysql (13)
    • pgsql (2)
    • redis (1)
    • tdengine (4)
  • 未分类 (7)
  • 程序员网赚 (20)
    • 广告联盟 (3)
    • 私域流量 (5)
    • 自媒体 (5)
  • 量化投资 (4)
  • 面试 (14)

功能

  • 登录
  • 文章RSS
  • 评论RSS
  • WordPress.org

All Rights Reserved by Gitweixin.本站收集网友上传代码, 如有侵犯版权,请发邮件联系yiyuyos@gmail.com删除.