【Ubuntu datasophon1.2.1 二开之九：验证离线数据入湖】

Ubuntu datasophon1.2.1 二开之九验证离线数据入湖背景环境准备1. 在datasophon安装好dolphinscheduler 3.1.8配置租户创建环境修改配置文件2. 升级spark3版本遇到坑及填平方法1.现象: 经典的 NoClassDefFoundError例如 org/apache/spark/kafka010/KafkaConfigUpdater 和 org/apache/spark/sql/connector/write/Write。2. Spark与Paimon版本不兼容3. HDFS权限问题4. 元数据存储方式选择5. 环境与组件升级6.Spark 找不到 Kafka 数据源下载匹配版本的 Kafka 依赖在 DS Spark 任务的选项参数中添加 --jars7.Paimon 表创建失败Derby 权限问题8.Kafka Topic 不存在查看所有 topic使用正确的 topic 名称9.Spark 版本不匹配导致类找不到10.DolphinScheduler 参数换行问题11.Shell 脚本换行符问题转换换行符或直接在 DS 的脚本框中编写不上传文件12.Shell 脚本中 Spark 命令缺少配置13.CSV 导入 ClickHouse 格式错误将 CSV 转换为 TSV逗号 → 制表符14.Shell 脚本 while 循环只执行一次不推荐while 循环逐行插入推荐管道批量导入15. Paimon 表写入失败表不存在最后背景前面一篇已经验证在线数据入湖了最后一步就是验证离线数据入湖。虽然已经做好坑坑洼洼的准备。但是困难比预想要多得多。花时间要多得多。为了最后一哆嗦坚持坚持坚持终于胜利环境准备1. 在datasophon安装好dolphinscheduler 3.1.8配置租户hdfs就是linux用户创建环境告诉ds spark3位置,hdfs位置修改配置文件修改bin\env\dolphinscheduler_env.sh总结一下1.修改数据库类型为mysql,指定数据库url用户名及密码2.修改zk地址3.修改hadoop位置及它配置目录4.jdk位置5.hive及flink位置修改 work-server\bin\start.sh就修改一行原来$JAVA_HOME/bin/java $JAVA_OPTS\-cp$DOLPHINSCHEDULER_HOME/conf:$DOLPHINSCHEDULER_HOME/libs/*\ org.apache.dolphinscheduler.server.worker.WorkerServer改成$JAVA_HOME/bin/java $JAVA_OPTS\-cp$DOLPHINSCHEDULER_HOME/conf:$DOLPHINSCHEDULER_HOME/libs/*:${HADOOP_CONF_DIR}:${HADOOP_HOME}/share/hadoop/common/*:${HADOOP_HOME}/share/hadoop/hdfs/* \ org.apache.dolphinscheduler.server.worker.WorkerServer启动时传入hdfs配置目录及home目录2. 升级spark3版本datasophon 自带spark3.1.3 ,不支持paimon换句话paimon支持spark 从3.2版本开始。必须升级否则报类找不到这个问题耽搁我好长时间paimon从0.7-0.9都试还试了kafka-connect方式。遇到坑及填平方法1.现象: 经典的 NoClassDefFoundError例如 org/apache/spark/kafka010/KafkaConfigUpdater 和 org/apache/spark/sql/connector/write/Write。填平方案:依赖分析: 通过错误栈和GitHub issue确认KafkaConfigUpdater 类属于 Spark 而非 Paimon且需要额外的 spark-token-provider-kafka 等JAR包。版本匹配: 最终放弃在Spark 3.1.3上挣扎将Spark升级到与Paimon 0.9.0兼容的3.2.4版本从根本上解决了 Write 类找不到的API不兼容问题。依赖管理: 明确需要通过 --jars 参数或DataSophon资源中心将 paimon-spark-3.2-0.9.0.jar 等所有依赖JAR包完整地提供给Spark任务。2. Spark与Paimon版本不兼容现象: 使用Paimon 0.8.2或0.9.0时均报告 Write 类找不到。填平方案:放弃低版本组合: 确认Paimon从某个版本开始强依赖Spark 3.2的DataSource V2 API与你的Spark 3.1.3环境不兼容。升级Spark: 决定将Spark从 3.1.3 升级到 3.2.4。这一步是解决问题的关键转折点虽然需要升级组件但一劳永逸地解决了兼容性问题。3. HDFS权限问题现象: 建表或写入时报 Permission denied用户 root123 无法在 /user/paimon/warehouse 或 /user/hive/warehouse 下创建目录。填平方案:使用 sudo -u hdfs: 切换到HDFS超级用户执行权限修复命令。递归授权: 执行 hdfs dfs -chown -R root123:supergroup /user/paimon 和 hdfs dfs -chmod -R 755 /user/paimon将目录所有者改为当前用户并赋予写权限。调整Hive仓库权限: 类似地对 /user/hive/warehouse 目录也进行了权限调整确保了元数据操作顺畅。4. 元数据存储方式选择现象: 使用Paimon默认的文件系统Catalog时元数据直接存在HDFS上导致并发问题和权限困扰。填平方案:切换为Hive Metastore: 最终决定使用 Hive Metastore 作为Paimon的Catalog。配置如下text–conf spark.sql.catalog.paimon.metastorehive–conf spark.sql.catalog.paimon.urithrift://ddp1:9083明确数据存储路径: 同时指定 spark.sql.catalog.paimon.warehouse实现元数据在MySQL/Hive与数据文件在HDFS的分离使整个架构更清晰、更稳定。5. 环境与组件升级现象: 旧版Spark 3.1.3成为兼容性瓶颈且DataSophon管理着多个组件升级有顾虑。填平方案:确认兼容性: 检查并确认Hadoop 3.3.3、Hive 3.1.3、Kafka 2.4.1等核心组件与Spark 3.2.4和JDK 11兼容。分步升级: 选择保持JDK 8不变仅升级Spark到3.2.4将影响面降到最低成功绕过JDK升级的风险。6.Spark 找不到 Kafka 数据源错误信息textFailed to find data source: kafka原因 Spark 默认不包含 Kafka 集成包。解决方案bash下载匹配版本的 Kafka 依赖spark-sql-kafka-0-10_2.12-3.2.0.jarkafka-clients-2.8.0.jarcommons-pool2-2.11.1.jar在 DS Spark 任务的选项参数中添加 --jars–jars /path/to/spark-sql-kafka-0-10_2.12-3.2.0.jar,/path/to/kafka-clients-2.8.0.jar,/path/to/commons-pool2-2.11.1.jar关键点版本必须与 Spark 版本匹配commons-pool2 是 Kafka 消费者的传递依赖容易遗漏7.Paimon 表创建失败Derby 权限问题错误信息textERROR XBM0H: Directory /…/metastore_db cannot be created.java.io.FileNotFoundException: derby.log (Permission denied)原因 Spark 默认使用嵌入式 Derby 作为 Hive Metastore没有配置 Hive Metastore 时会尝试在临时目录创建 Derby 数据库。解决方案 配置使用外部的 Hive Metastorebash–conf spark.sql.catalog.paimon.metastorehive–conf spark.sql.catalog.paimon.urithrift://hive-metastore-host:9083–conf spark.sql.extensionsorg.apache.paimon.spark.extensions.PaimonSparkSessionExtensions8.Kafka Topic 不存在错误信息textUNKNOWN_TOPIC_OR_PARTITION原因 脚本中配置的 topic 名称与实际不符。解决方案bash查看所有 topickafka-topics.sh --bootstrap-server kafka-host:9092 --list使用正确的 topic 名称kafka_topic “user_log_topic” # 而不是 “user_log”9.Spark 版本不匹配导致类找不到错误信息textjava.lang.NoClassDefFoundError: org/apache/spark/kafka010/KafkaConfigUpdaterjava.lang.ClassNotFoundException: org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig原因 Spark 版本与 Kafka 依赖版本不匹配。解决方案Spark 3.2.0 → 使用 spark-sql-kafka-0-10_2.12-3.2.0.jar 和 kafka-clients-2.8.0.jar不能混用 3.2.4 的包10.DolphinScheduler 参数换行问题错误信息text–conf: command not found原因 DS 的选项参数中换行导致命令被分割。解决方案 所有参数写在一行不要换行properties–jars /path/to/jar1.jar,/path/to/jar2.jar --conf key1value1 --conf key2value211.Shell 脚本换行符问题错误信息text$‘\r’: command not found原因 在 Windows 上编辑的脚本上传后带有 \r\n 换行符。解决方案bash转换换行符sed -i ‘s/\r$//’ script.sh或直接在 DS 的脚本框中编写不上传文件12.Shell 脚本中 Spark 命令缺少配置错误信息textERROR XBM0H: Directory /…/metastore_db cannot be created.原因 Shell 脚本中的 spark-sql 命令没有配置 Hive Metastore。解决方案 在 Shell 脚本中的每个 spark-sql 命令都要加上完整配置bash${SPARK_HOME}/bin/spark-sql–conf spark.sql.catalog.paimonorg.apache.paimon.spark.SparkCatalog–conf spark.sql.catalog.paimon.metastorehive–conf spark.sql.catalog.paimon.urithrift://ddp1:9083–conf spark.sql.catalog.paimon.warehousehdfs://…/warehouse–conf spark.sql.extensionsorg.apache.paimon.spark.extensions.PaimonSparkSessionExtensions13.CSV 导入 ClickHouse 格式错误错误信息textCode: 117. DB::Exception: Expected end of line: … (INCORRECT_DATA)原因CSV 中的 JSON 字段包含逗号和引号干扰了 CSV 解析时间格式 2026-03-29T10:00:00.000Z 不是 ClickHouse 默认格式解决方案 使用 TSV 格式替代 CSVbash将 CSV 转换为 TSV逗号 → 制表符cat data.csv | sed ‘s/,/\t/g’ | clickhouse client --query “INSERT INTO table FORMAT TSV”14.Shell 脚本 while 循环只执行一次错误信息 循环内的命令导致循环提前退出。原因 clickhouse-client 命令失败或 echo 干扰了循环变量。解决方案 放弃 while 循环使用管道直接导入bash不推荐while 循环逐行插入while read line; doclickhouse-client --query “INSERT …”done file.csv推荐管道批量导入cat file.csv | clickhouse-client --query “INSERT INTO table FORMAT TSV”15. Paimon 表写入失败表不存在错误信息textSchema file not found in location paimon.default.ods_user_log. Please create table first.解决方案 先创建表再写入数据sqlCREATE TABLE IF NOT EXISTS paimon.default.ods_user_log (…) USING paimon;或在 Python 脚本中自动创建pythonif not spark.catalog.tableExists(“paimon.default.ods_user_log”):spark.sql(“CREATE TABLE …”)最后成果截图各个节点对应脚本1.kafka_to_paimonfrom pyspark.sqlimportSparkSession from pyspark.sql.functionsimportcol,lit,from_json,to_timestamp from pyspark.sql.typesimportStructType,StructField,StringType,TimestampTypeimportsysimportos bizdatesys.argv[1]iflen(sys.argv)1elseos.environ.get(bizdate,2026-03-29)sparkSparkSession.builder \.appName(fKafkaToPaimon_Batch_{bizdate})\.config(spark.sql.catalog.paimon,org.apache.paimon.spark.SparkCatalog)\.config(spark.sql.catalog.paimon.warehouse,hdfs://nameservice1/user/paimon/warehouse)\.config(spark.sql.catalog.paimon.metastore,hive)\.config(spark.sql.catalog.paimon.uri,thrift://ddp1:9083)\.config(spark.sql.extensions,org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions)\.getOrCreate()print(f开始处理业务日期: {bizdate})# 定义 schema json_schemaStructType([StructField(user_id,StringType(),True),StructField(event_time,StringType(),True),StructField(event_type,StringType(),True),StructField(data,StringType(),True)])# Kafka 配置 kafka_bootstrap_serversddp4:9092,ddp3:9092kafka_topicuser_log_topic# 检查表是否存在如果不存在则创建print(检查表是否存在...)try:# 尝试查询表 spark.sql(SELECT 1 FROM paimon.default.ods_user_log LIMIT 1)print(表已存在)except Exception:print(表不存在正在创建...)create_sqlCREATETABLEpaimon.default.ods_user_log(user_idSTRING,event_timeTIMESTAMP,event_typeSTRING,dataSTRING,dtSTRING)USINGpaimonPARTITIONEDBY(dt)TBLPROPERTIES(bucket4,file.formatparquet) spark.sql(create_sql)print(表创建成功)# 从 Kafka 读取数据 dfspark.read \.format(kafka)\.option(kafka.bootstrap.servers,kafka_bootstrap_servers)\.option(subscribe,kafka_topic)\.option(startingOffsets,earliest)\.option(endingOffsets,latest)\.load()\.selectExpr(CAST(value AS STRING) as json_str)# 解析JSONparsed_dfdf.select(from_json(col(json_str),json_schema).alias(data)).select(col(data.user_id),to_timestamp(col(data.event_time)).alias(event_time),col(data.event_type),col(data.data),lit(bizdate).alias(dt)).filter(col(user_id).isNotNull())countparsed_df.count()print(f从 Kafka 读取到 {count} 条有效数据)ifcount0:print(\n数据样例前5条)parsed_df.show(5,truncateFalse)# 写入数据 parsed_df.write \.format(paimon)\.mode(append)\.insertInto(paimon.default.ods_user_log)print(f✅ 写入完成共 {count} 条记录)else:print(没有数据需要写入)spark.stop()2.paimon写入ck(实际是写入hdfs,忘记改名称from pyspark.sqlimportSparkSession bizdate2026-03-29sparkSparkSession.builder \.appName(fPaimonToClickHouse_{bizdate})\.config(spark.sql.catalog.paimon,org.apache.paimon.spark.SparkCatalog)\.config(spark.sql.catalog.paimon.metastore,hive)\.config(spark.sql.catalog.paimon.uri,thrift://ddp1:9083)\.config(spark.sql.catalog.paimon.warehouse,hdfs://nameservice1/user/paimon/warehouse)\.config(spark.sql.extensions,org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions)\.getOrCreate()print(*50)print(Paimon → ClickHouse 导入)print(f业务日期: {bizdate})print(*50)# 读取数据 dfspark.sql(fSELECTuser_id,event_time,event_type,dataFROMpaimon.default.ods_user_logWHEREdt{bizdate})countdf.count()print(f[1/2] Paimon 表中数据量: {count})ifcount0:print([2/2] 写入 ClickHouse...)# 保存为CSV到HDFSoutput_pathf/tmp/paimon_export_{bizdate}df.coalesce(1).write.mode(overwrite).option(header,false).csv(output_path)print(f✅ 数据已导出到 HDFS: {output_path})print(f 请使用 clickhouse-client 导入数据)else:print(⚠️ 没有数据需要导入)spark.stop()3.hdfs导入ck#!/bin/bash# 环境变量CH_CLIENT/opt/datasophon/clickhouse/bin/clickhouseHOSTddp3PORT9000USERdefaultBIZDATE2026-03-29echoechoPaimon → ClickHouse 数据导入echo业务日期: ${BIZDATE}echo#1.下载CSV文件 echo[1/5] 下载 CSV 文件...hdfs dfs-get-f/tmp/paimon_export_${BIZDATE}/part-*.csv/tmp/paimon_data_${BIZDATE}.csv2/dev/nullif[!-f/tmp/paimon_data_${BIZDATE}.csv];then echo❌ CSV 文件不存在exit1fiTOTAL_LINES$(wc-l/tmp/paimon_data_${BIZDATE}.csv)echoCSV 文件行数: ${TOTAL_LINES}#2.创建 ClickHouse 表如果不存在 echo[2/5] 创建 ClickHouse 表如果不存在...${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--query CREATETABLEIFNOTEXISTSdefault.user_log(user_id String,event_time String,event_type String,data String)ENGINEMergeTree()ORDERBYuser_id 2/dev/null#3.删除当前日期的旧数据避免重复 echo[3/5] 删除当前日期的旧数据...${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--query ALTERTABLEdefault.user_logDELETEWHEREevent_timeLIKE${BIZDATE}%2/dev/null#4.导入数据使用TSV格式 echo[4/5] 导入数据到 ClickHouse...# 将CSV转换为TSV用制表符分隔后导入 cat/tmp/paimon_data_${BIZDATE}.csv|seds/,/\t/g|${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--queryINSERT INTO default.user_log FORMAT TSV#5.验证导入结果 echo[5/5] 验证导入结果...COUNT$(${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--querySELECT COUNT(*) FROM default.user_log WHERE event_time LIKE ${BIZDATE}%2/dev/null)TOTAL$(${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--querySELECT COUNT(*) FROM default.user_log2/dev/null)echoecho✅ 导入完成echo 本次导入行数: ${TOTAL_LINES}echo 当前日期数据量: ${COUNT}echo ClickHouse 总数据量: ${TOTAL}echo# 显示数据样例if[${TOTAL_LINES}-gt0];then echoecho新增数据样例:${CH_CLIENT}client--host ${HOST}--port ${PORT}--user ${USER}--querySELECT * FROM default.user_log WHERE event_time LIKE ${BIZDATE}% LIMIT 3--format PrettyCompact fi # 清理临时文件 rm-f/tmp/paimon_data_${BIZDATE}.csv echoechoecho导入任务完成echo如需沟通lita2lz