Ceph分布式集群安装配置

2022-09-29

前言

记录Ceph分布式集群安装配置过程及问题解决

1. 服务器环境说明

主机名	角色	IP地址
ceph1	ceph-deploy、mon、mgr、osd	192.168.44.128
ceph2	mon、mgr、osd	192.168.44.129
ceph3	mon、mgr、osd	192.168.44.130

要保证每台机器能连接外网

2. 服务器环境准备

2.1 配置yum源

wget -qO /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-7.repo
wget -qO /etc/yum.repos.d/epel.repo http://mirrors.aliyun.com/repo/epel-7.repo

vi /etc/yum.repos.d/ceph.repo
[ceph]
name=ceph
baseurl=https://mirrors.aliyun.com/ceph/rpm-luminous/el7/x86_64/
gpgcheck=0

[ceph-noarch]
name=cephnoarch
baseurl=https://mirrors.aliyun.com/ceph/rpm-luminous/el7/noarch/
gpgcheck=0

more >>

展开全文 >>

Flink SQL 客户端查询Hive配置及问题解决

2022-08-26

前言

记录博主第一次使用Flink SQL查询Hive的配置以及问题解决过程

博主目前还没有用过Flink,没有写过Flink代码，完全是一个小白。之所以使用Flink，是因为博主目前需要测试使用dolphinscheduler的一些功能，其中包括Flink数据源，这里Flink数据源我们使用的是Kyuubi Flink Engine,且学会、了解了Flink之后对于学习Hudi、Kyuubi也是有帮助的。

版本

flink-1.14.3，这里使用kyuubi1.5.2自带的Flink

环境

HDP环境，Hadoop、Hive等已经安装配置好

配置

HADOOP_CLASSPATH

1	export HADOOP_CLASSPATH=`hadoop classpath`

可以添加在比如/ect/profile里，这样等于修改全局的环境，如果想只对Flink生效，可以添加在bin/config.sh文件里

一开始我对hadoop classpath理解错了，以为是自己手动修改成实际的路径，但是hadoop对应的jar包的路径有很多，后来发现它的意思是执行命令hadoop classpath将返回值赋给HADOOP_CLASSPATH

more >>

展开全文 >>

Hudi Spark SQL源码学习总结-select（查询）

2022-08-15

前言

上一篇文章Hudi Spark源码学习总结-spark.read.format(“hudi”).load分析了load方法直接查询Hudi表路径的源码逻辑，那么Spark SQL select 表名的方式和load最终走的逻辑是一样的吗？本文带着这个疑问来分析一下select查询Hudi表的源码逻辑

版本

Spark 2.4.4
Hudi master 0.12.0-SNAPSHOT 最新代码
（可以借助Spark3 planChangeLog 打印日志信息查看哪些规则生效,Spark3和Spark2生效的规则大致是一样的）

示例代码

先Create Table，再Insert几条数据

1	spark.sql(s"select id, name, price, ts, dt from $tableName").show()

more >>

展开全文 >>

Hudi Spark源码学习总结-spark.read.format("hudi").load（2）

2022-08-12

前言

补充上一篇文章Hudi Spark源码学习总结-spark.read.format(“hudi”).load，由于上篇文章篇幅已经比较长了，所以单独写一篇补充一下，没有读过的可以先阅读一下，我们在上篇文章讲到resolveBaseFileOnlyRelation返回的是HadoopFsRelation，那么如果返回BaseFileOnlyRelation呢?

起因

其实除了HadoopFsRelation、BaseFileOnlyRelation还有IncrementalRelation、MergeOnReadSnapshotRelation、MergeOnReadIncrementalRelation、HoodieBootstrapRelation，之所有要单独看一下BaseFileOnlyRelation，是因为我们从提交历史上可以看出，一开始默认的就是HadoopFsRelation后来添加了自定义的BaseFileOnlyRelation，然后现在回退成了HadoopFsRelation，查看对应PR了解了一下原因

添加BaseFileOnlyRelation的PR：[HUDI-3338] custom relation instead of HadoopFsRelation，原因是：目前，HadoopFsRelation将使用实际分区路径的值作为分区字段的值。但是，与普通表不同，Hudi在Parquet文件中保留分区值。在某些情况下，实际分区路径的值和分区字段的值是不同的。因此，这里实现了BaseFileOnlyViewRelation，通过它，Hudi可以管理自己的关系。
回退为HadoopFsRelation的PR：[HUDI-3902] Fallback to HadoopFsRelation in cases non-involving Schema Evolution,原因是：Spark的一些优化规则（以及一些其他处理）是基于HadoopFsRelation的使用，这导致当我们依赖自定义Relation时，这些优化没有得到应用。我们可以在[HUDI-3338]的PR的下面可以看到，有commiter提出，使用BaseFileOnlyRelation后查询性能相比于之前的HadoopFsRelation下降了大约一倍，这是因为Spark仅在少数地方做了查询优化，其中一个地方就是我们在上篇文章中讲到的FileSourceStrategy,我们看到[HUDI-3902]虽然回退到HadoopFsRelation，但是貌似并没有处理分区路径值的问题，所以后面又提了一个PR：
PR：[HUDI-3204] Fixing partition-values being derived from partition-path instead of source columns，这个PR添加了自定义FileFormat:Spark24HoodieParquetFileFormat，它重写了ParquetFileFormat的buildReaderWithPartitionValues方法。其中的关键点是添加了参数shouldAppendPartitionValues来判断是否将分区路径中的值添加为分区字段的值。

if (shouldAppendPartitionValues) {
  vectorizedReader.initBatch(partitionSchema, file.partitionValues)
} else {
  vectorizedReader.initBatch(StructType(Nil), InternalRow.empty)
}

more >>

展开全文 >>

Hudi Spark源码学习总结-spark.read.format("hudi").load

2022-08-11

前言

由于工作原因，之前查询Hudi主要是用Hive来查询的，所以对Hive查询Hudi的逻辑比较了解，但是对于Spark查询Hudi的逻辑不太了解。所以现在想要学习一下Spark查询Hudi的大概逻辑，搞清楚它是如何从Spark的源码跳转到Hudi源码执行Hudi查询的逻辑，这样既能搞清楚Spark查询表的逻辑，也能搞清楚Spark查询Hudi的逻辑，也便于在后面使用Kyuubi Spark SQL 时出现问题能更好的定位解决。

版本

Spark 2.4.4
Hudi master 0.12.0-SNAPSHOT 最新代码
（可以借助Spark3 planChangeLog 打印日志信息查看哪些规则生效）

示例代码

先用上篇文章写Hudi数据，再进行查询

import spark.implicits._
val df = Seq((1, "a1", 10, 1000)).toDF("id", "name", "value", "ts")
df.write.format("hudi")
  .option(HoodieWriteConfig.TBL_NAME.key, tableName)
  .option(TABLE_TYPE.key, COW_TABLE_TYPE_OPT_VAL)
  .option(RECORDKEY_FIELD.key, "id")
  .option(PRECOMBINE_FIELD.key, "ts")
  .option(PARTITIONPATH_FIELD.key, "")
  .option(KEYGENERATOR_CLASS_NAME.key, classOf[NonpartitionedKeyGenerator].getName)
  .option(HoodieWriteConfig.INSERT_PARALLELISM_VALUE.key, "1")
  .option(HoodieWriteConfig.UPSERT_PARALLELISM_VALUE.key, "1")
  .mode(SaveMode.Overwrite)
  .save(tmp.getCanonicalPath)

spark.read.format("hudi").load(tmp.getCanonicalPath).show()

more >>

展开全文 >>