Presto Hive MetaStore相关代码分析

云原生数据湖 2020-04-29

661

Presto是一个Connector只能连接到一个数据源，比如Hive Connector，一个Hive Connector底层只对应一个Hive Metastore，如果公司内部有两个Hive Metastore, 你就需要搞两个 hive.properties
以加载两个不同的Hive Connector来服务不同的Hive Metastore, 如果有三个甚至更多的Hive Metastore呢？通过加载新的不同的 hive.properties
来解决这个问题从Presto的架构角度来看是不错的方式，因为它体现了Presto的插件方式的成功，但是对于最终使用者来说其实是不友好的，能否一个Hive Connector来服务多个Hive Metastore呢？今天带着这个问题来分析一下Presto里面Hive Metastore相关的代码。

Metastore相关的代码结构

Presto为了支持各种不同的底层 Hive MetaStore的实现，相关的代码结构搞得比较复杂，相关类图如下:

其中 SemiTransactionalHiveMetastore
是最终 Presto 用到的 Metastore，而为了支持底层各种不同的Metastore的实现，比如标准的Hive Metastore, 或者AWS的Glue，或者用于测试的FileMetaStore，而抽象了 ExtendedHiveMetastore
这一层。

在 ExtendedHiveMetastore
之外还有一个 HiveMetastore
的接口抽象，这两个接口很大部分是一样的。但是有一些细微的差别，比如 ExtendedHiveMetastore
里面会有一些“很好用”的方法比如:


void replaceTable(String databaseName, String tableName, 
                  Table newTable, PrincipalPrivileges principalPrivileges);
void renameTable(String databaseName, String tableName, 
                 String newDatabaseName, String newTableName);
void addColumn(String databaseName, String tableName, 
               String columnName, HiveType columnType, String columnComment);
void renameColumn(String databaseName, String tableName, 
                  String oldColumnName, String newColumnName);

而这些方法对应到 HiveMetastore
接口里面只有一个方法: alterTable
。而 ExtendedHiveMetastore
的实现类: BridgingHiveMetastore
在中间做了适配从而使得Presto Hive Connector用起 ExtendedHiveMetastore
很好用，也就是说 ExtendedHiveMetastore
是面向 Presto 引擎的接口，而 HiveMetastore
是面向底层 Apache Hive Metastore
的接口。

这里另外一个值得一说的是 SemiTransactionalHiveMetastore
, 相对于底下的各个Metastore类只是对接、适配各种外界不同的元数据服务的实现，这个类实现了Presto在操作数据 + 元数据过程中的各种”事务“性的操作，我们知道底层接口只提供了关于元数据的各种单个的接口。多个接口之间是没有事务的概念的，但是某些数据操作是需要某种程度的事务的，比如Insert Overwrite, 要实现一个可用的Insert Overwrite，它的过程大概是这样的:

所谓的Insert Overwrite是指对于一个已经有数据的表tbl001, 我们新插入一些数据，把老的数据覆盖掉。我们这里假定tbl001表数据是保存在HDFS上的，对应的路径是: hdfs://bucket001/tbl001/

把要插入的数据先写入HDFS的一个临时目录: hdfs://tmp/a/
把老的数据备份到一个临时目录: hdfs://bucket001/tbl001_backup
把新的数据move到最终目录: hdfs://tmp/a/
-> hdfs://bucket001/tbl001/
如果一切顺利我们需要删除掉老的数据目录: hdfs://bucket001/tbl001_backup
如果上面任何一步失败的，我们需要把老的数据恢复: hdfs://bucket001/tbl001_backup
-> hdfs://bucket001/tbl001/
完成元数据层面的变更:

dropTable(oldTableDefinition)
createTable(newTableDefinition)

这里 SemiTransactionalMetastore
配合 HiveMetadata
完成上面的一些列操作，当然这里也没有实现绝对的事务性，因此这个类的名字上有一个 Semi
。

用户信息从引擎到 SemiTransactionalHiveMetastore

为了能够让一个Hive Connector支持多个底层的Hive Metastore, 我们必须要让 SemiTransactionalHiveMetastore
知道当前请求的用户是谁，这个 用户
在Presto 里面就是 ConnectorIdentity
, 而 ConnectorIdentity
来自于 Session
对象，也就是说我们要让 SemiTransactionalHiveMetastore
感受到当前的 Session
。构造(new) SemiTransactionalHiveMetaStore
对象的代码路径是:


// 这条线会构造一个 TransactionMetadata 出来
MetadataManager#getCatalogMetadata(Session session, ConnectorId connectorId)
  - TransactionManager.getCatalogMetadata -- 这里没有传session了，丢掉了
    - InMemoryTransactionManager#getCatalogMetadata
      - InMemoryTransactionManager#getTransactionMetadata -- 这里获取了 TransactionMetadata.


// 这条线从 TransactionMetadata 最终创建SemiTransactionalMetastore出来
TransactionMetadata#getTransactionCatalogMetadata -- 这里会最终调用到Connector#beginTransaction
- TransactionMetadata#createConnectorTransactionMetadata
- TransactionMetadata#beginTransaction
- HiveConnector#beginTransaction
- HiveConnectorFactory#create
- HiveMetadataFactory#get

因此为了要让 HiveMetadataFactory#get
在创建 SemiTransactionalMetastore
的时候能够获得 ConnectorIdentity
，我们需要在创建 TransactionalMetadata
的时候把 ConnectorIdentity
传进去，那么 TransactionalMetadata
的构造路径是怎么样的呢?

QueryStateMachine#beginWithTicker(...Session session)
- TransactionManager#beginTransaction -- 这个产生TransactionMetadata

可以看到我们只要把 ConnectorIdentity
传递到 TransactionManager#beginTransaction
即可。

用户信息从引擎到HDFS

前面分析了把用户信息从引擎传递到Metastore的代码路径，这只解决了元数据部分的事情，不同的Hive Metastore底层往往对应了不同的HDFS集群往往有不同的配置，至少要把当前用户的信息传递到HDFS的配置(org.apache.hadoop.conf.Configuration)里面去，这个点目前Presto就已经支持，我们可以注册一个: DynamicConfigurationProvider
, 比如社区为了支持Google Cloud Storage，使用了下面的扩展方式:


public class GcsConfigurationProvider
        implements DynamicConfigurationProvider
{
    private static final String GCS_OAUTH_KEY = "hive.gcs.oauth";


    @Override
    public void updateConfiguration(Configuration configuration, HdfsContext context, URI uri)
{
        if (!uri.getScheme().equals(GoogleCloudStorageFileSystem.SCHEME)) {
            return;
        }


        String accessToken = context.getIdentity().getExtraCredentials().get(GCS_OAUTH_KEY);
        if (accessToken != null) {
            configuration.set(GCS_ACCESS_TOKEN_CONF, accessToken);
        }
    }
}


后记
数据湖开发者社区由 阿里云开发者社区 与 阿里云Data Lake Analytics团队 共同发起，致力于推广数据湖相关技术，包括hudi、delta、spark、presto、oss、元数据、存储加速、格式发现等，学习如何构建数据湖分析系统，打造适合业务的数据架构。扫描下方钉钉群二维码，加入社区一起学习讨论。
阿里云Data Lake Analytics是Serverless化的交互式联邦查询服务。使用标准SQL即可轻松分析与集成对象存储(OSS)、数据库(PostgreSQL/MySQL等)、NoSQL(TableStore等)数据源的数据。
Data Lake Analytics产品详情页：https://www.aliyun.com/product/datalakeanalytics
Data Lake Analytics 1元购入口：https://common-buy.aliyun.com/?commodityCode=openanalytics_post#/buy

数据库

文章转载自云原生数据湖，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Presto Hive MetaStore相关代码分析

Metastore相关的代码结构

用户信息从引擎到 SemiTransactionalHiveMetastore

用户信息从引擎到HDFS

后记

评论