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");
//這個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)用了上面的這個函數(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());
//會讀取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)
{
//省略
//重點是這句話
addCatalogConnector(catalogName, connectorId, connectorFactory, properties);
//省略
}
private synchronized void addCatalogConnector(String catalogName, ConnectorId connectorId, ConnectorFactory factory, Map<String, String> properties)
{
//創(chuàng)建一個connector缰儿，由之前的factorys來獲取HiveConnectorFactory（presto/spi/connector包下）
Connector connector = createConnector(connectorId, factory, properties);
//標準的表畦粮，也就是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的動作：
1、將connector放入到map中
2乖阵、獲取ConnectorSplitManager（spi具體connector實現(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);
}
#具體的實現(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ù)據(jù)泛鸟。待續(xù)~
