1、從hive中拉數(shù)據(jù)

1.1 <b>入口</b>：在presto指定根目錄下/etc/catalog/hive.properties

1.2 <b>讀取配置文件</b>com.facebook.presto.connector.ConnectorManager：

private void loadCatalog(File file)
            throws Exception
    {
        //讀取catalog下面的文件，把后綴名給去掉
        String catalogName = Files.getNameWithoutExtension(file.getName());
        if (disabledCatalogs.contains(catalogName)) {
            log.info("Skipping disabled catalog %s", catalogName);
            return;
        }

        log.info("-- Loading catalog %s --", file);
        //加載文件中的配置信息到map中
        Map<String, String> properties = new HashMap<>(loadProperties(file));

        //獲取connector.name 這里一般是寫hive不同版本（hive-hadoop2等）
        String connectorName = properties.remove("connector.name");
        checkState(connectorName != null, "Catalog configuration %s does not contain connector.name", file.getAbsoluteFile());

        //創(chuàng)建連接
        connectorManager.createConnection(catalogName, connectorName, ImmutableMap.copyOf(properties));
        log.info("-- Added catalog %s using connector %s --", catalogName, connectorName);
    }

<b>1.3 創(chuàng)建連接到hive</b>

//catalogName=hive connectorName=我使用的是hive-hadoop2  properties包含hive.metastore.uri連接
connectorManager.createConnection(catalogName, connectorName, ImmutableMap.copyOf(properties));

public synchronized ConnectorId createConnection(String catalogName, String connectorName, Map<String, String> properties)
    {
        requireNonNull(connectorName, "connectorName is null");
        //這個(gè)connectorFactories怎么來的?
        ConnectorFactory connectorFactory = connectorFactories.get(connectorName);
        checkArgument(connectorFactory != null, "No factory for connector %s", connectorName);
        return createConnection(catalogName, connectorFactory, properties);
    }

疑問：上面的connectorFactories是怎么來的呢？

 public synchronized void addConnectorFactory(ConnectorFactory connectorFactory)
    {
        ConnectorFactory existingConnectorFactory = connectorFactories.putIfAbsent(connectorFactory.getName(), connectorFactory);
    }

誰調(diào)用了上面的這個(gè)函數(shù)呢？
<b>在PluginManager類中有這樣的函數(shù)：

 public void installPlugin(Plugin plugin)
    {
        ......
        for (com.facebook.presto.spi.ConnectorFactory connectorFactory : plugin.getLegacyConnectorFactories()) {
            log.info("Registering legacy connector %s", connectorFactory.getName());
            //會(huì)讀取plugin的目錄，然后加載插件，添加到factory中
            connectorManager.addConnectorFactory(new LegacyTransactionConnectorFactory(connectorFactory));
        }

    public PluginManager(
            參數(shù)省略)
    {
         ......
        //讀取配置信息獲取plugin所在的目錄地址，即服務(wù)的plugin
        installedPluginsDir = config.getInstalledPluginsDir();
        if (config.getPlugins() == null) {
            this.plugins = ImmutableList.of();
        }
        else {
            this.plugins = ImmutableList.copyOf(config.getPlugins());
        }

 public void loadPlugins()
            throws Exception
    {
      ....
        //讀取plugin目錄下的所有目錄加載插件，并加載
        for (File file : listFiles(installedPluginsDir)) {
            if (file.isDirectory()) {
                loadPlugin(file.getAbsolutePath());
            }
        }

        for (String plugin : plugins) {
            loadPlugin(plugin);
        }
.......
    }

<b>1.4 來看看具體的加載數(shù)據(jù)</b>

上面我們已經(jīng)明確了創(chuàng)建連接的過程了，接下面要介紹如何讀取hive中的數(shù)據(jù)。

我們回到上面createConnection的方法

  private synchronized ConnectorId createConnection(String catalogName, ConnectorFactory connectorFactory, Map<String, String> properties)
    {
        //省略 
        //重點(diǎn)是這句話
        addCatalogConnector(catalogName, connectorId, connectorFactory, properties);
        //省略
    }

    private synchronized void addCatalogConnector(String catalogName, ConnectorId connectorId, ConnectorFactory factory, Map<String, String> properties)
    {
        //創(chuàng)建一個(gè)connector，由之前的factorys來獲取HiveConnectorFactory（presto/spi/connector包下）
        Connector connector = createConnector(connectorId, factory, properties);

        //標(biāo)準(zhǔn)的表，也就是hive中的用戶表信息
        addConnectorInternal(ConnectorType.STANDARD, catalogName, connectorId, connector);

        //hive的表結(jié)構(gòu)信息
        ConnectorId informationSchemaId = createInformationSchemaConnectorId(connectorId);
        addConnectorInternal(ConnectorType.INFORMATION_SCHEMA, catalogName, informationSchemaId, new InformationSchemaConnector(catalogName, nodeManager, metadataManager));

        //系統(tǒng)表信息
        ConnectorId systemId = createSystemTablesConnectorId(connectorId);
        addConnectorInternal(ConnectorType.SYSTEM, catalogName, systemId, new SystemConnector(
                systemId,
                nodeManager,
                connector.getSystemTables(),
                transactionId -> transactionManager.getConnectorTransaction(transactionId, connectorId)));
    }

其中addConnectorInternal的動(dòng)作：

1、將connector放入到map中
2、獲取ConnectorSplitManager（spi具體connector實(shí)現(xiàn)）
3、獲取PageSourceProvider
4、獲取PageSinkProvider
5、獲取IndexProvider
6、獲取NodePartitioningProvider
7、獲取AccessControl
8、把上面獲取到的對象添加到presto自身的Manager中，比如PageSourceProvider添加到PageSourceManager管理中。

<b>我們關(guān)心的從哪里加載數(shù)據(jù)，應(yīng)該是PageSourceProvider作用</b>

#PageSourceManager類
 @Override
    public ConnectorPageSource createPageSource(Session session, Split split, List<ColumnHandle> columns)
    {
        getPageSourceProvider(split).createPageSource(split.getTransactionHandle(), connectorSession, split.getConnectorSplit(), columns);
    }

#具體的實(shí)現(xiàn)類HivePageSourceProvider
  @Override
    public ConnectorPageSource createPageSource(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, List<ColumnHandle> columns)
    {
        //讀取列信息
        List<HiveColumnHandle> hiveColumns = columns.stream()
                .map(HiveColumnHandle::toHiveColumnHandle)
                .collect(toList());

        //獲取hiveSplit路徑
        HiveSplit hiveSplit = checkType(split, HiveSplit.class, "split");
        Path path = new Path(hiveSplit.getPath());
        //hive的數(shù)據(jù)來源
        Optional<ConnectorPageSource> pageSource = createHivePageSource(
                cursorProviders,
                pageSourceFactories,
                hiveSplit.getClientId(),
                hdfsEnvironment.getConfiguration(path),
                session,
                path,
                hiveSplit.getStart(),
                hiveSplit.getLength(),
                hiveSplit.getSchema(),
                hiveSplit.getEffectivePredicate(),
                hiveColumns,
                hiveSplit.getPartitionKeys(),
                hiveStorageTimeZone,
                typeManager);
        if (pageSource.isPresent()) {
            return pageSource.get();
        }
        throw new RuntimeException("Could not find a file reader for split " + hiveSplit);
    }

好了，我們已經(jīng)知道了他是如何獲取數(shù)據(jù)的了，下一節(jié)介紹什么時(shí)候去拉取數(shù)據(jù)。待續(xù)~