Vertica系列:Vertica和Hadoop的互操作性

Vertica 8和 Hadoop 集群的互操作性已經很不錯的, 但這塊肯定是Vertica研發的重點, 將來可能還有較大的變動.

Vertica 集群 和 Hadoop 集群的兩種布局方式

集群布局	描述	許可證	支持Vertica for SQL on Hadoop特性
共享集群	Vertica 節點部署在 Hadoop 所有節點或部分節點上	Vertica for SQL on Hadoop 許可證	支持
分離集群	Vertica 集群和 Hadoop 分屬不同的集群	社區版或Premium版許可證	不支持

從目前情況下, 還是推薦使用分離模式, 主要原因有:

• 共存模式需要為 yarn 和 Vertica 分別分配合適的計算資源, 操作難度較大.
• 存儲方面 Vertica 推薦采用 Raid10 或 Raid50, 而 HDFS 不推薦使用 Raid.
• Hadoop Balancer功能很可能會影響Vertica的執行效率(通常情況下Vertica僅僅部署在集群中的幾台).

但長期看, 共享集群應該是Vertica重點的發展方向, HAWQ 就是這個思路. 從目前看到的資料, Vertica 9將有很大的進展.

Vertica 和 Hadoop 互操作方式匯總

方法	描述	推薦	效率	必須共享集群嗎
Reading Directly from HDFS	以External table 或 bulk load的方式讀取HDFS上的格式化文件	Recommended	高	無需
HCatalog Connector	使用Vertica SQL的方式訪問 Hive 的數據表		中低	無需
HDFS Connector	以External table 或 bulk load的方式讀取HDFS上的格式化文件	Deprecated	高	無需
ROS Containers on HDFS	在HDFS上存放Vertica ROS文件		高	必須
MapReduce Connector	使用MapReduce或Pig來訪問 HDFS 上的數據		中/低	?
Export to Parquet	導出Parquet文件到本地或HDFS上			無需
ParallelExport	導出文件到本地或HDFS上			無需

讀取原生 Hadoop 文件格式

在官網中好像叫做Reading Directly from HDFS 方式 或 HDFS Schema 方式.
該方法是用來替代 HDFS Connector 方法. 在使用該方法之前, Vertica 節點應能訪問 Hadoop 配置文件:

• 如果共享集群, Vertica 自然能訪問到這些配置文件, 只需要在 Vertica DB 中驗證 HadoopConfDir 環境變量即可.
• 如果分離集群, Vertica 必須將所需文件復制到這些節點, 但一個更好的做法是, 將 Vertica 節點配置為 Hadoop 的邊緣節點(英文為 edge node). 從 Hadoop 的角度來看，Vertica 就是一款客戶端應用程序. 可以使用 Hortonworks Ambari 或 Cloudera Manager 配置邊緣節點.

在 Vertica DB 中設置 HadoopConfDir 參數, 其路徑應該包括一個含有 core-site.xml 和 hdfs-site.xml 文件的目錄.
ALTER DATABASE mydb SET HadoopConfDir = '/hadoop/hcat/conf';

驗證配置是否正確:
select VERIFY_HADOOP_CONF_DIR( )

該訪問方式效率較高,但需要注意的是,

• ORC 或 Parquet 文件不得使用復雜數據類型.
• 由 Hive 或 Impala 壓縮的文件需要進行 Zlib (GZIP) 或 Snappy 壓縮, 不支持 LZO 壓縮.
• 定義外部表時，您必須定義文件中的所有列.

使用 hdfs 方案不會消除對 webHDFS 服務訪問權限的需求, hdfs 方案並不適用於所有的文件. 如果 hdfs 不可用，那么 Vertica 會自動改用 webhdfs. 可以直接使用 webhdfs:// 前綴，並指定主機名、端口和文件路徑, 讀取不受 Kerberos 保護的文件時，使用 hdfs 方案可能會提供更好的性能.

[推薦] 使用 Hadoop NameService 的方式訪問 hdfs 上的數據, 注意是兩個 slash 的寫法, hadoopNS 這個name service 已經在 hdfs-site.xml 定義好了.
CREATE EXTERNAL TABLE tt (a1 INT, a2 VARCHAR(20))
AS COPY FROM 'hdfs://hadoopNS/data/file.csv';

[不推薦] 如果沒有在hdfs-site.xml中定義name service, 需要使用 hdfs:/// 這樣的寫法.
CREATE EXTERNAL TABLE tt (a1 INT, a2 VARCHAR(20))
AS COPY FROM 'hdfs:///opt/data/file.csv';

直接支持 ORC/Parquet 格式,
=> CREATE EXTERNAL TABLE tableName (columns)
AS COPY FROM path ORC;
=> CREATE EXTERNAL TABLE tableName (columns)
AS COPY FROM path PARQUET;
=> CREATE EXTERNAL TABLE t (id int, name varchar(50),
created date, region varchar(50))
AS COPY FROM 'hdfs:///path///*'
PARQUET(hive_partition_cols='created,region');

直接 HDFS 的 bulk load 示例:
COPY t FROM 'hdfs:///opt/data/file1.dat';

如果您在啟動 Vertica 后更新了任何Hadoop的配置文件，使用下面的語句強制更新 Vertica 中作更新.
SELECT CLEAR_HDFS_CACHES ();

HCatalog Connector 訪問方式

首先Vertica 集群需要安裝 Hadoop and hive的 jar library, 同時能拿到正確的 Hadoop 和 Hive 的配置文件(hive-site.xml/core-site.xml/yarn-site.xml/webhcat-site.xml/hdfs-site.xml).
Vertica 專門提供了 hcatUtil 工具幫助我們完成這些配置, 該工具的位置是: /opt/vertica/packages/hcat/tools/
詳細使用見 https://my.vertica.com/docs/8.1.x/HTML/index.htm#Authoring/HadoopIntegrationGuide/HCatalogConnector/ConfiguringVerticaForHCatalog.htm

另外, 還可以配置是要使用 HiveServer2 還是 WebHCat
ALTER DATABASE mydb SET HCatalogConnectorUseHiveServer2 = 1; -- 使用 HiveServer2, 缺省模式.
ALTER DATABASE mydb SET HCatalogConnectorUseHiveServer2 = 0; -- 使用 WebHCat, 效率比 HiveServer2 好.

集群Linux 完成安裝后, 還需要在 Vertica DB 中安裝必要的Udx庫, Vertica 8.1缺省未安裝.
首先通過下面SQL驗證 VHCatalogLib user_lib 是否已經安裝?
select * from user_library_manifest where user_library_manifest.lib_name = 'VHCatalogLib';
如果沒有安裝的化, 需要先安裝, vsql -f /opt/vertica/packages/hcat/ddl/install.sql

創建 HCatalog schema 名為 hcat
CREATE HCATALOG SCHEMA hcat WITH
hostname='hcathost'
HCATALOG_SCHEMA='default'
HCATALOG_USER='hcatuser';

可以使用 hcat schema直接訪問 hive 表了.
SELECT
distinct ship_type,
ship_mode,
ship_carrier
FROM hcat.shipping_dimension
WHERE shipping_key >= 10627
GROUP BY ship_mode,
ship_carrier;

HDFS Connector 的外部表訪問示例

首先通過下面SQL驗證 HdfsSource user_lib 是否已經安裝? Vertica 8.1缺省已經安裝.
select * from user_library_manifest where user_library_manifest.lib_name = 'HdfsSource';
如果沒有安裝的化, 需要先安裝, vsql -f /opt/vertica/packages/hdfsconnector/ddl/install.sql

CREATE EXTERNAL TABLE shipping_dimension
(shipping_key integer,
ship_type char(30),
ship_mode char(10),
ship_carrier char(20))
AS COPY FROM SOURCE
Hdfs(url='http://n01:50070/webhdfs/v1/dbadmin/shippingDimension/*');

HDFS Connector 的 bulk load 示例:
COPY testTable SOURCE Hdfs(url='http://hadoop:50070/webhdfs/v1/tmp/test.txt', username='hadoopUser');

ROS Containers on HDFS 方式

共享集群暫時不做過多研究, 需要創建 HDFS 存儲位置
=> CREATE LOCATION 'webhdfs://hadoop:50070/user/dbadmin' ALL NODES SHARED USAGE 'data' LABEL 'coldstorage';
=> SELECT node_name,location_path,location_label FROM STORAGE_LOCATIONS;

-- Vertica Proprietary Format in HDFS
=> SELECT set_object_storage_policy('DBNAME','HDFS');

導出到HDFS

EXPORT TO PARQUET 可以支持本地導出, 也可以導出到HDFS上.
EXPORT TO PARQUET(directory='hdfs:///data/sales_data')
AS SELECT * FROM public.sales;

EXPORT TO PARQUET(directory='hdfs:///data/sales_by_region')
AS SELECT sale.price, sale.date, store.region
FROM public.sales sale
JOIN public.vendor store ON sale.distribID = store.ID;

EXPORT TO PARQUET(directory = 'hdfs:///data/export')
OVER(PARTITION BY date) AS SELECT date, price FROM public.sales;

EXPORT TO PARQUET(directory = 'hdfs:///data/export')
OVER(PARTITION BY date ORDER BY price) AS SELECT date, price FROM public.sales;

使用第三方庫 ParallelExport 導出到HDFS

EXPORT TO PARQUET 的導出格式必須是 Parquet, 第三方函數 ParallelExport 支持很多中導出格式, 但效率比EXPORT TO PARQUET要差.
參考< http://www.dbjungle.com/exporting-vast-amounts-of-data-using-parallel-export-for-hpe-vertica/>

select ParallelExport(eutime,eid::varchar,logintype::varchar,sessiontime
using parameters cmd='hadoop fs -put - /user/etl/app/app_fact_event_base/${nodeName}', separator=' '
) over( partition auto ) from app.app_fact_event_base where date(sessiontime) in ('2016-12-13','2016-12-14','2017-02-12','2017-02-13') limit 1000;