Apache CarbonData 里程碑式版本 1.3 发布

  • 陈亮

2018 年 2 月 8 日

话题:Apache语言 & 开发AI

看新闻很累?看技术新闻更累?试试下载 InfoQ 手机客户端,每天上下班路上听新闻,有趣还有料!

CarbonData 是一种新的高性能数据存储格式,已在 20+ 企业生产环境上部署和使应用,企业数据规模达到万亿。针对当前大数据领域分析场景需求各异而导致的存储冗余问题,业务驱动下的数据分析灵活性要求越来越高,CarbonData 提供了一种新的融合数据存储方案,以一份数据同时支持多种应用场景,并通过多级索引、字典编码、预聚合、动态 Partition、准实时数据查询、列存等特性提升了 IO 扫描和计算性能,实现百亿数据级秒级响应。

我们来看下,CarbonData 1.3.0 有哪些重大特性:

1. 支持与 Spark 2.2.1 集成

CarbonData 1.3.0 支持与最新稳定的 Spark 2.2.1 版本集成。

2. 支持预聚合表特性

在 1.3.0 中,CarbonData 的预聚合特性,与传统 BI 系统的 CUBE 方案最大区别是,用户不需要改任何 SQL 语句,既可加速 group by 的统计性能,又可查询明细数据,做到一份数据满足多种应用场景。具体的用法如下:

a) 创建主表:

CREATE TABLE sales (
order_time TIMESTAMP,
user_id STRING,
sex STRING,
country STRING,
quantity INT,
price BIGINT)
STORED BY 'carbondata'

b) 基于上面主表 sales 创建预聚合表:

CREATE DATAMAP agg_sales
ON TABLE sales
USING "preaggregate"
AS
SELECT country, sex, sum(quantity), avg(price)
FROM sales
GROUP BY country, sex

c) 用户不需要改 SQL 语句,基于主表 sales 的查询语句如命中预聚合表 agg_sales,可以显著提升查询性能:

SELECT country, sex, sum(quantity), avg(price) FROM sales GROUP BY country, sex;// 命中,完全和聚合表一样 

SELECT sex, sum(quantity) FROM sales GROUP BY sex;// 命中,聚合表的部分查询 

SELECT country, avg(price) FROM sales GROUP BY country;// 命中,聚合表的部分查询 

SELECT country, sum(price) FROM sales GROUP BY country;// 命中,因为聚合表里 avg(price) 是通过 sum(price)/count(price) 产生,所以 sum(price) 也命中 

SELECT sex, avg(quantity) FROM sales GROUP BY sex; // 没命中,需要创建新的预聚合表 

SELECT max(price), country FROM sales GROUP BY country;// 没命中,需要创建新的预聚合表 

SELECT user_id, country, sex, sum(quantity), avg(price) FROM sales GROUP BY user_id, country, sex; // 没命中,需要创建新的预聚合表 

d) 在 3.0 版本中,支持的预聚合表达式有:SUM、AVG、MAX、MIN、COUNT

e) 实测性能可提升 10+ 倍以上,大家可以参考例子,把测试数据调到 1 亿规模以上,跑下这个例子:/apache/carbondata/examples/PreAggregateTableExample.scala

3. 支持时间维度的预聚合特性,并支持自动上卷

此特性为 Alpha 特性,当前时间粒度支持设置为 1,比如:支持按 1 天聚合,暂不支持指定 3 天,5 天的粒度进行聚合,下个版本将支持。支持自动上卷(Year,Month,Day,Hour,Minute),具体用法如下:

a) 创建主表:

CREATE TABLE sales (
order_time TIMESTAMP,
user_id STRING,
sex STRING,
country STRING,
quantity INT,
price BIGINT)
STORED BY 'carbondata'

b) 分别创建 Year、Month、Day、Hour、Minute 粒度的聚合表:

CREATE DATAMAP agg_year
ON TABLE sales
USING "timeseries"
DMPROPERTIES (
'event_time’=’order_time’,
'year_granualrity’=’1’,
) AS
SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
 avg(price) FROM sales GROUP BY order_time, country, sex

CREATE DATAMAP agg_month
ON TABLE sales
USING "timeseries"
DMPROPERTIES (
'event_time’=’order_time’,
'month_granualrity’=’1’,
) AS
SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
 avg(price) FROM sales GROUP BY order_time, country, sex

CREATE DATAMAP agg_day
  ON TABLE sales
  USING "timeseries"
  DMPROPERTIES (
  'event_time’=’order_time’,
  'day_granualrity’=’1’, // 当前粒度只支持设置为 1,
  ) AS
  SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
   avg(price) FROM sales GROUP BY order_time, country, sex

CREATE DATAMAP agg_sales_hour
ON TABLE sales
USING "timeseries"
DMPROPERTIES (
'event_time’=’order_time’,
'hour_granualrity’=’1’,
) AS
SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
 avg(price) FROM sales GROUP BY order_time, country, sex

CREATE DATAMAP agg_minute
ON TABLE sales
USING "timeseries"
DMPROPERTIES (
'event_time’=’order_time’,
'minute_granualrity’=’1’,
) AS
SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
 avg(price) FROM sales GROUP BY order_time, country, sex

c) 用户可不用创建所有时间粒度的聚合表,系统支持自动 roll-up 上卷,如:已创建了 Day 粒度的聚合表,当查询 Year、Month 粒度的 group by 聚合时,系统会基于已聚合好的 Day 粒度值推算出 Year、Month 粒度的聚合值:

CREATE DATAMAP agg_day
  ON TABLE sales
  USING "timeseries"
  DMPROPERTIES (
  'event_time’=’order_time’,
  'day_granualrity’=’1’,
  ) AS
  SELECT order_time, country, sex, sum(quantity), max(quantity), count(user_id), sum(price),
   avg(price) FROM sales GROUP BY order_time, country, sex

(Year、Month 粒度的聚合查询,可用上面创建的 agg_day 上卷)

SELECT timeseries(order_time, ‘month’), sum(quantity) FROM sales group by timeseries(order_time,
  ’month’)
SELECT timeseries(order_time, ‘year’), sum(quantity) FROM sales group by timeseries(order_time,
  ’year’)

4. 支持实时入库,准实时查询

在 1.3.0 中,支持通过 Structured Streaming 实时导入数据到 CarbonData 表,并立即可查询这些 fresh 数据。

a) 实时获取数据:

val readSocketDF = spark.readStream
   .format("socket")
   .option("host", "localhost")
   .option("port", 9099)
   .load()

b) 写数据到 CarbonData 表

qry = readSocketDF.writeStream
   .format("carbondata")
   .trigger(ProcessingTime("5 seconds"))
   .option("checkpointLocation", tablePath.getStreamingCheckpointDir)
   .option("dbName", "default")
   .option("tableName", "carbon_table")
   .start()

(具体可参考例子 /apache/carbondata/examples/CarbonStructuredStreamingExample.scala)

5. 支持标准的 Partition 特性:

此 Partition 和 Hive 和 Spark partition 一样,用户可以按字段值建立 partition 分区,查询时可指定具体分区数据进行快速查询;与 SORT_COLUMNS 组合应用,可以建立多级排序,满足任意维度组合的过滤查询,做到一份数据满足多种应用场景。如:创建下面表,设置 productDate 作为 partition 字段,数据按天进行分区;再通过 SORT_COLUMNS 建立多维 MDK 索引。这样可以按照 productDate,productName, storeProvince, storeCity 任意过滤组合快速查询数据。

CREATE TABLE IF NOT EXISTS productSalesTable (
productName STRING,
  storeProvince STRING,
storeCity STRING,
  saleQuantity INT,
  revenue INT)
PARTITIONED BY (productDate DATE)
STORED BY 'carbondata'
TBLPROPERTIES(‘SORT_COLUMNS’ = ‘productName, storeProvince, storeCity’)

6. 支持 CREATE TABLE AS SELECT 语法

CREATE TABLE carbon_table STORED BY 'carbondata' AS SELECT * FROM parquet_table

7. 支持指定导入的数据进行查询

CarbonData 每批次导入的数据,会放到一个 segment 下,在 1.3.0 里用户可以指定 segment 数据进行查询,即:用户可以指定数据批次按需查询。

a) 查询 Segment ID 列表

SHOW SEGMENTS FOR TABLE <databasename>.<table_name> 

b) 设置 Segment ID

SET carbon.input.segments.<databasename>.<table_name> = <list of segment IDs>

(具体可参考例子:/apache/carbondata/examples/QuerySegmentExample.scala)

8. Apache CarbonData官网:apache.org

1.3.0下载地址

Apache语言 & 开发AI