Spark 2.1.0与CarbonData 1.0.0集群模式部署及使用入门指南

Size: px

Start display at page:

Download "Spark 2.1.0与CarbonData 1.0.0集群模式部署及使用入门指南"

夯汛范
5 years ago
Views:

1 Spark 与 CarbonData 集群模式部署及使用入门指南本文作者 : 李寅威, 从事大数据机器学习方面的工作, 目前就职于 CVTE 联系方式 : 微信 (coridc), 邮箱 ( @qq.com) 原文链接 : Spark CarbonData1.0.0 集群模式部署及使用入门 1 引言 Apache CarbonData 是一个面向大数据平台的基于索引的列式数据格式, 由华为大数据团队贡献给 Apache 社区, 目前最新版本是版介于目前主流大数据组件应用场景的局限性,Car bondata 诞生的初衷是希望通过仅保存一份数据来满足不同的应用场景, 如 : OLAP 顺序存取 (Sequential Access) 随机存取 (Random Access) CarbonData 也被评为 2016 年的 BLACKDUCK 奖, 有关 CarbonData 的相关资料如下 : 官网地址 : Github: Mailing list: dev@carbondata.incubator.apache.org cwiki: Jira 地址 : 本文主要介绍 Spark CarbonData1.0.0 集群模式部署流程, 并辅以一个小案例来讲解如何在 Spark shell 下使用 CarbonData 2 准备工作 2.1 集群规划 id hostname mem cpu storage 1 master 32G Intel(R) Core(TM) i GHz 2 slave1 32G Intel(R) Core(TM) i GHz SATA3 7200RPM 4T SATA3 7200RPM 8T 1 / 21

2 id hostname mem cpu storage 3 slave2 32G Intel(R) Core(TM) i GHz 4 slave3 32G Intel(R) Core(TM) i GHz 5 slave4 32G Intel(R) Core(TM) i GHz SATA3 7200RPM 8T SATA3 7200RPM 8T SATA3 7200RPM 8T 2.2 系统环境操作系统下载地址 : 建议版本 :Unix-like environment (Linux, Mac OS X) 版本查看 : # 示例 (CentOS) [hadoop@master ~]$ cat /etc/redhat-release CentOS release 6.8 (Final) JDK 下载地址 : 建议版本 :JDK 版本查看 : [hadoop@master ~]$ java -version java version "1.8.0_60" Git 下载地址 : 建议版本 : 无版本查看 : 2 / 21

3 ~]$ git --version git version Maven 下载地址 : 建议版本 :3.0.4 版本查看 : [hadoop@master ~]$ mvn -v Apache Maven (bb52d8502b132ec0a5a3f4c09453c dc5; T00:41:47+0 8:00) Maven home: /opt/maven Hadoop 下载地址 : 建议版本 :2.7.2 版本查看 : [hadoop@master ~]$ hadoop version Hadoop [hadoop@master ~]$ echo $HADOOP_HOME /opt/hadoop Scala 下载地址 : 建议版本 :2.11.x 版本查看 : [hadoop@master ~]$ scala -version Scala code runner version Copyright , LAMP/EPFL 3 / 21

4 Spark 下载地址 : 建议版本 :2.1.0 部署模式 :Standalone/YARN 版本查看 : [hadoop@master spark-2.1.0]$./bin/spark-submit --version Welcome to / / / / \ \/ _ \/ _ `/ / '_/ / /. /\_,_/_/ /_/\_\ version /_/ Using Scala version , Java HotSpot(TM) 64-Bit Server VM, 1.8.0_60 [hadoop@master ~]$ echo $SPARK_HOME /opt/spark Thrift 下载地址 : 建议版本 :0.9.3 版本查看 : [hadoop@master ~]$ thrift -version Thrift version 编译及部署 3.1 编译 Step 1: 源码下载 $ git clone carbondata 4 / 21

5 Step 2: 修改 Maven 私有仓库地址 ( 可选 ) 由于网络原因, 从 Maven 中央仓库下载 jar 包可能非常慢, 大家可根据自己的实际情况修改为企业内部私有仓库或阿里云等外部源, 如 : # 修改 conf/setting.xml 文件 <mirrors> <mirror> <id>nexus</id> <name>nexus</name> <url> <mirrorof>*</mirrorof> </mirror> </mirrors> Step 3: 编译打包 [hadoop@master ~]$ cd carbondata [hadoop@master carbondata]$ mvn clean package -DskipTests -Pspark-2.1 -Dspark.version= Phadoop 在编译打包的过程中,maven 会自动下载所依赖的 jar 包, 但可能还会有部分 jar 包无法下载成功导致打包失败的情况, 此时需要我们手动去网上下载并将对应的 jar 包放到 Maven localrepository 的对应目录下并重新执行上述命令, 执行成功后, 会出现以下提示 : [INFO] [INFO] Reactor Summary: [INFO] [INFO] Apache CarbonData :: Parent... SUCCESS [ s] [INFO] Apache CarbonData :: Common... SUCCESS [16:82 min] [INFO] Apache CarbonData :: Core... SUCCESS [03:23 min] [INFO] Apache CarbonData :: Processing... SUCCESS [ s] [INFO] Apache CarbonData :: Hadoop... SUCCESS [ s] [INFO] Apache CarbonData :: Spark Common... SUCCESS [ s] [INFO] Apache CarbonData :: Spark2... SUCCESS [ s] [INFO] Apache CarbonData :: Spark Common Test... SUCCESS [ s] [INFO] Apache CarbonData :: Assembly... SUCCESS [ s] [INFO] Apache CarbonData :: Spark2 Examples... SUCCESS [ s] [INFO] [INFO] BUILD SUCCESS 5 / 21

6 [INFO] [INFO] Total time: 29:42 min [INFO] Finished at: T09:26:38+08:00 [INFO] Final Memory: 211M/2513M [INFO] 集群模式部署注 : 本节是基于 Spark Standalone 模式的配置部署,Spark on Yarn 模式下的配置部署类似但不完全相同, 具体可参照文档 : - Installing and Configuring CarbonData on Standalone Spark Cluster - Installing and Configuring CarbonData on Spark on YARN Cluster 部署步骤如下 : Step 1: 复制 CarbonData jar 包 # 创建目录并拷贝 [hadoop@master spark-2.1.0]$ mkdir $SPARK_HOME/carbonlib [hadoop@master spark-2.1.0]$ cp -r ~/carbondata/assembly/target/scala-2.11/carbondata_ incubating-shade-hadoop2.7.2.jar./carbonlib # 编辑环境变量 [hadoop@master ~]$ vi /opt/spark-2.1.0/conf/spark-env.sh # 添加以下配置 export SPARK_CLASSPATH=$SPARK_CLASSPATH:${SPARK_HOME/carbonlib/* Step 2: 配置 carbon.properties # 拷贝至 spark conf 目录 [hadoop@master ~]$ cp ~/carbondata/conf/carbon.properties.template /opt/spark-2.1.0/conf/ carbon.properties # 修改配置 #Mandatory. Carbon Store path carbon.storelocation=hdfs://master:9000/opt/carbonstore #Base directory for Data files carbon.ddl.base.hdfs.url=hdfs://master:9000/opt/data 6 / 21

7 #lock 文件存储方式 carbon.lock.type=hdfslock Step 3: 配置 spark-defaults.conf ~]$ vi /opt/spark-2.1.0/conf/spark-defaults.conf # 添加以下配置 spark.executor.extrajavaoptions on.properties spark.driver.extrajavaoptions n.properties -Dcarbon.properties.filepath=/opt/spark-2.1.0/conf/carb -Dcarbon.properties.filepath=/opt/spark-2.1.0/conf/carbo Step 4: 分发至集群各节点 # Spark - conf [hadoop@master ~]$ scp -r /opt/spark-2.1.0/conf/ hadoop@slave1:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/conf/ hadoop@slave2:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/conf/ hadoop@slave3:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/conf/ hadoop@slave4:/opt/spark-2.1.0/ # Spark - carbonlib [hadoop@master ~]$ scp -r /opt/spark-2.1.0/carbonlib/ hadoop@slave1:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/carbonlib/ hadoop@slave2:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/carbonlib/ hadoop@slave3:/opt/spark-2.1.0/ [hadoop@master ~]$ scp -r /opt/spark-2.1.0/carbonlib/ hadoop@slave4:/opt/spark-2.1.0/ 4 数据准备 4.1 数据描述为更聚焦于整个流程, 我们的数据集仅含一张表, 共 21 个字段, 详细信息如下 : 7 / 21

8 ID COLUMN TYPE CARDINALITY COMMENT 1 id BIGINT TOTAL_NUM ID 2 order_code STRING TOTAL_NUM 订单编号 3 sales_area_id INT 100 销售区域 ID 4 sales_id INT 销售人员 ID 5 order_inputer INT 100 录单人员 ID 6 pro_type STRING 1000 产品型号 7 currency INT 50 订单币种 8 exchange_rate DECIMAL 1000 当前汇率 9 unit_cost_price DECIMAL 成本单价 10 unit_selling_pric e DECIMAL 销售单价 11 order_num INTEGER 1000 订单数量 12 order_amount DECIMAL / 订单金额 13 order_discount DOUBLE 9 订单折扣 14 order_account_a mount DECIMAL / 实际金额 15 order_time TIMESTAMP 下单时间 16 delivery_channel INT 80 发货渠道 ID 17 delivery_address STRING 发货地址 18 recipients STRING 收件人 19 contact STRING 联系方式 20 delivery_date DATE 发货日期 21 comments STRING 备注 4.2 数据生成我们编写一段程序来解决数据批量生成的问题, 为方便有兴趣的童鞋直接使用, 所有代码都写在一个 Java 类中, 相关配置 ( 如数据量等 ) 可直接在代码中修改当然大家也完全可以用自己的数据进行测试数据生成的代码如下 ( 注 : 需使用 JDK1.8 及以上版本编译运行 ): package com.cvte.bigdata.pt; 8 / 21

9 import java.io.ioexception; import java.nio.file.files; import java.nio.file.paths; import java.nio.file.standardopenoption; import java.time.localdate; import java.time.localdatetime; import java.time.format.datetimeformatter; import java.util.uuid; import java.util.concurrent.arrayblockingqueue; public class BatchDataGenerate implements Runnable { ); // total record private final Integer TOTAL_NUM = 1 * * 10000; // batch size of writing to file and echo infos to console private final Integer BATCH_SIZE = 10 * 10000; // capacity of the array blocking queue private final Integer QUEUE_CAPACITY = 100 * 10000; // csv file path that save the data private final String DEST_FILE_PATH = "E:\\order_detail.csv"; // the queue which the write thread write into and the read thread read from private ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY // the cardinalities of the fields(dimensions) which means the distinct // number of a column private final int CARDINALITY_SALES_AREA_ID = 100; private final int CARDINALITY_SALES_ID = 10000; private final int CARDINALITY_ORDER_INPUTER = 100; private final int CARDINALITY_PRO_TYPE = 1000; private final int CARDINALITY_CURRENCY = 50; private final int CARDINALITY_EXCHANGE_RATE = 1000; private final int CARDINALITY_UNIT_COST_PRICE = 10000; private final int CARDINALITY_UNIT_SELLING_PRICE = 10000; private final int CARDINALITY_ORDER_NUM = 1000; private final int CARDINALITY_ORDER_DISCOUNT = 9; private final int CARDINALITY_ORDER_TIME = ; private final int CARDINALITY_DELIVERY_CHANNEL = 80; private final int CARDINALITY_DELIVERY_ADDRESS = ; private final int CARDINALITY_RECIPIENTS = ; private final int CARDINALITY_CONTACT = ; private final int CARDINALITY_DELIVERY_DATE = 10000; private final int CARDINALITY_COMMENTS = 9 / 21

10 public void run() { try { if ("tgenerate".equals(thread.currentthread().getname())) { // data generating thread generatedata(); else if ("twrite".equals(thread.currentthread().getname())) { // data writing thread savedatatofile(); catch (InterruptedException e) { e.printstacktrace(); catch (IOException e) { e.printstacktrace(); /** generating data for table order_details and called by thread InterruptedException void liyinwei 2017 年 3 月 9 日下午 7:19:16 */ private void generatedata() throws InterruptedException { for (int i = 0; i < TOTAL_NUM; i++) { StringBuffer sb = new StringBuffer(); sb.append(i + 1) // 2. 订单编号.append(",").append(UUID.randomUUID()) // 3. 销售区域 ID.append(",").append((i+3) % CARDINALITY_SALES_AREA_ID) // 4. 销售人员 ID.append(",").append((i + 4) % CARDINALITY_SALES_ID) // 5. 录单人员 ID.append(",").append((i + 5) % CARDINALITY_ORDER_INPUTER) // 6. 产品型号.append(",").append("PRO_TYPE_" + (i + 6) % CARDINALITY_PRO_TYPE) // 7. 订单币种.append(",").append((i + 7) % CARDINALITY_CURRENCY) // 8. 当前汇率.append(",").append((i + 8) % CARDINALITY_EXCHANGE_RATE) // 9. 成本单价.append(",").append((i + 9) % CARDINALITY_UNIT_COST_PRICE) // 10. 销售单价.append(",").append((i + 10) % CARDINALITY_UNIT_SELLING_PRICE) 10 / 21

11 // 11. 订单数量.append(",").append((i + 11) % CARDINALITY_ORDER_NUM) // 12. 订单金额.append(",").append(((i + 10) % CARDINALITY_UNIT_SELLING_PRICE) * ((i + 11) % CA RDINALITY_ORDER_NUM)) // 13. 订单折扣.append(",").append(String.format("%.2f", (i + 13) % CARDINALITY_ORDER_DISCOU NT * 0.1)) // 14. 实际金额.append(",").append(String.format("%.2f", ((i + 10) % CARDINALITY_UNIT_SELLING_PRICE) * ((i + 11) % CARDINALITY_OR DER_NUM) * ((i + 13) % CARDINALITY_ORDER_DISCOUNT * 0.1))) // 15. 下单时间.append(",").append(LocalDateTime.now().plusSeconds(i % CARDINALITY_ORDER_TIME).format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss"))) // 16. 发货渠道 ID.append(",").append((i + 16) % CARDINALITY_DELIVERY_CHANNEL) // 17. 发货地址.append(",").append("DELIVERY_ADDRESS_" + (i + 17) % CARDINALITY_DELIVERY_AD DRESS) // 18. 收件人.append(",").append("RECIPIENTS_" + (i + 18) % CARDINALITY_RECIPIENTS) // 19. 联系方式.append(",").append( l + (i + 19) % CARDINALITY_CONTACT) // 20. 发货日期.append(",").append(LocalDate.now().plusDays(i % CARDINALITY_DELIVERY_DATE)) // 21. 备注.append(",").append("RECIPIENTS_" + (i + 21) % CARDINALITY_COMMENTS); queue.put(sb.tostring()); if (i % BATCH_SIZE == 0) { System.out.println(i + " records have generated successfully."); System.out.println("current queue length is: " + queue.size()); /** writing data from array block queue to file and called by thread IOException void 11 / 21

12 liyinwei 2017 年 3 月 9 日下午 8:16:59 */ private void savedatatofile() throws InterruptedException, IOException { int i = 0; StringBuffer sb = new StringBuffer(); while (true) { sb.append(queue.take()).append("\n"); i++; if (i % BATCH_SIZE == 0) { Files.write(Paths.get(DEST_FILE_PATH), sb.tostring().getbytes(), StandardOpenOption.CREATE, StandardOpenOption.APPEND); sb.setlength(0); System.out.println(i + " records have written to file successfully."); System.out.println("current queue length is: " + queue.size()); public static void main(string[] args) { BatchDataGenerate batchdatagenerate = new BatchDataGenerate(); // data generating thread Thread thread1 = new Thread(batchDataGenerate, "tgenerate"); // data writing thread Thread thread2 = new Thread(batchDataGenerate, "twrite"); thread1.start(); thread2.start(); 以上代码的逻辑比较简单, 构造两个线程, 一个用于生成数据并保存在队列中, 另一个用于从队列中取数据并保存在文件中代码中几个关键的变量 ( 参数 ) 介绍如下 : TOTAL_NUM: 需要生成的数据总条数 BATCH_SIZE: 控制台打印进度信息及队列当前信息的数据条数间隔, 同时也是每批次写文件的数据量大小 QUEUE_CAPACITY: 队列大小, 若队列满则阻塞写线程 DEST_FILE_PATH: 数据文件存储目录 queue: 缓存队列 12 / 21

13 运行上述代码, 稍等片刻, 我们就可以在目标路径中找到 order_detail.csv 文件,3 亿条数据占用 6 1.8G 存储空间 4.3 数据上传接着, 我们将 order_detail.csv 上传至 HDFS 中 : [hadoop@master ~]$ hadoop fs -put /home/hadoop/order_detail.csv /data/carbondata/pt 5 启动 spark-shell 在使用 CarbonData 前, 有以下先决条件 : 5.1 简介 Hadoop HDFS 及 YARN 已安装及启动 Spark 已安装并在每个节点启动 CarbonData 用户拥有读写 HDFS 的权限 CarbonData 对 Spark 已有较好支持, 我们可将 carbondata 的 jar 包传入 sparkshell 来使用 CarbonData:./bin/spark-shell --jars <carbondata assembly jar path> 进入上述 Shell 后, 会自动生成一个名为 spark 的 SparkSession 及一个名为 sc 的 SparkContext 5.2 案例样例代码如下 : [hadoop@master ~]$ cd /opt/spark [hadoop@master spark-2.1.0]$ carbondata_jar=./carbonlib/$(ls -1 carbonlib grep "^carbonda ta_.*\.jar$") [hadoop@master spark-2.1.0]$./bin/spark-shell --num-executors 4 --executor-cores 1 --drivermemory 5g --total-executor-cores 16 --executormemory 5g --master spark://master: jars ${carbondata_jar 13 / 21

14 细心的小伙伴可能已经发现, 在样例代码中, 我们在启动 spark shell 时额外添加了几个参数, 这里要强调一点的是,Spark 的配置及 spark shell 的配置会很大程度影响查询性能, 因此, 加入我们需要做组件性能的横向对比, 需注意以下事项 : 1. 保持参数的一致性, 强调单一变量原则 ; 2. 需为集群配置合理的参数, 以尽可能地发挥组件的性能 ; 关于 Spark 的配置可参照以下文档 : - Cluster Launch Scripts - Spark Standalone Mode - Launching Applications with spark-submit - Submitting Applications 6 创建 CarbonSession 6.1 简介接下来我们需要创建一个 CarbonSession 来对 CarbonData 相关的内容进行操作 : # import packages import org.apache.spark.sql.sparksession import org.apache.spark.sql.carbonsession._ # create CarbonSession val carbon = SparkSession.builder().config(sc.getConf).getOrCreateCarbonSession("<hdfs store path>") 6.2 案例 import org.apache.spark.sql.sparksession import org.apache.spark.sql.carbonsession._ val carbon = SparkSession.builder().config(sc.getConf).getOrCreateCarbonSession("hdfs://mast er:9000/opt/carbonstore") 至此, 就完成了 CarbonSession 的创建 7 创建数据库及数据表 7.1 简介关于 CarbonData 表的创建, 可参照 CarbonData 官方文档中的 CREATE TABLE 部分, 为充分发挥 CarbonData 的优势, 我们需注意以下内容 : 14 / 21

15 - 字典编码 (Dictionary Encoding) 默认情况下 CarbonData 会自动为 String 类型的字段创建字典, 而非 String 类型的字段不会创建字典我们可根据实际需求进行调整, 如取消对部分高基列的字典编码等, 以下为对应的语法 : TBLPROPERTIES ("DICTIONARY_EXCLUDE"="column1, column2") TBLPROPERTIES ("DICTIONAR Y_INCLUDE"="column1, column2") - 列组 (Column group) 对于经常一起出现的列, 我们可以将其设置为列组, 被设置为列组的列会按行的方式进行存储, 然后与其它列或列组一起按列存储默认情况下,CarbonData 不自动对列进行组合以下为对应的语法 : TBLPROPERTIES ("COLUMN_GROUPS"="(column1, column3), (Column4,Column5,Column6)") 块大小 (Table Block Size) CarbonData 支持 1M-2018M 大小的块, 默认为 1024M, 我们可按需进行调整, 以下为对应的语法 : TBLPROPERTIES ("TABLE_BLOCKSIZE"="512 MB") 倒排索引 (Inverted Index) 倒排索引在压缩率和查询效率方面有非常大的作用, 这对低基维度 ( 默认情况下,CarbonData 会将所有非数值类型的字段视为维度, 而将数值型字段视为度量值 ) 效果尤其明显默认情况下, CarbonData 会自动创建, 我们可根据实际需求进行调整, 如取消对部分高级维度的倒排索引以下为对应的语法 : TBLPROPERTIES ("NO_INVERTED_INDEX"="column1, column3") 7.2 案例 15 / 21

16 在本文的例子中, 我们可设计建表语句如下 : CREATE TABLE IF NOT EXISTS pt.order_detail ( id BIGINT, order_code STRING, sales_area_id INT, sales_id INT, order_inputer INT, pro_type STRING, currency INT, exchange_rate DECIMAL, unit_cost_price DECIMAL, unit_selling_price DECIMAL, order_num INTEGER, order_amount DECIMAL, order_discount DOUBLE, order_account_amount DECIMAL, order_time TIMESTAMP, delivery_channel INT, delivery_address STRING, recipients STRING, contact STRING, delivery_date DATE, comments STRING ) STORED BY 'carbondata' TBLPROPERTIES ( 'COLUMN_GROUPS' = '(recipients,contact)', 'DICTIONARY_EXCLUDE' = 'comments', 'DICTIONARY_INCLUDE' = 'sales_area_id,sales_id', 'NO_INVERTED_INDEX' = 'id,order_code' ) 在 spark-shell 中, 我们可通过以下命令建表 : carbon.sql("create TABLE IF NOT EXISTS pt.order_detail ( id BIGINT, order_code STRING, sales _area_id INT, sales_id INT, order_inputer INT, pro_type STRING, currency INT, exchange_rate D ECIMAL, unit_cost_price DECIMAL, unit_selling_price DECIMAL, order_num INTEGER, order_am ount DECIMAL, order_discount DOUBLE, order_account_amount DECIMAL, order_time TIMEST AMP, delivery_channel INT, delivery_address STRING, recipients STRING, contact STRING, deliv ery_date DATE, comments STRING ) STORED BY 'carbondata' TBLPROPERTIES ( 'COLUMN_GRO UPS' = '(recipients,contact)', 'DICTIONARY_EXCLUDE' = 'comments', 'DICTIONARY_INCLUDE' = 's ales_area_id,sales_id', 'NO_INVERTED_INDEX' = 'id,order_code' )") 16 / 21

17 8 数据查询 8.1 数据导入 CarbonData 支持两种方式的数据导入, 分别为 : - 直接通过 CSV 文件导入 CarbonData 表 - 通过 spark-sql API 导入导入方式一简介第一种数据导入方式是将原始数据从 CSV 文件导入 CarbonData 表形成 Carbon 格式的文件, 其语法为 : LOAD DATA [LOCAL] INPATH 'folder_path' INTO TABLE [db_name.]table_name OPTIONS(prope rty_name=property_value,...) 由上述命令中我们可以看到,CarbonData 支持从 LOCAL 及非 LOCAL( 如 HDFS) 位置导入,OPTI ONS 选项支持的内容可查看 DML Operations on CarbonData 需要注意的是,CarbonData 并不会自动根据顺序识别 CSV 文件中的列与数据表字段间的关系, 为此有两种解决方案, 第一种是在 CSV 文件的第一行加上列名, 与数据保持相同的分隔符, 如 : id,username,age 1,zhangsan,20 2,lisi,21 第二种是在 load data 命令中添加 fileheader, 如 : OPTIONS('FILEHEADER'='column1,column2') 案例 17 / 21

18 在 spark-shell 中, 我们可以用 benchmark{ 来进行性能测试, 使用 benchmark{ 需导入包 : import org.apache.spark.sql.catalyst.util._ 结合 benchmark, 我们一起来看一下以下的案例 : import org.apache.spark.sql.sparksession import org.apache.spark.sql.carbonsession._ import org.apache.spark.sql.catalyst.util._ val carbon = SparkSession.builder().config(sc.getConf).getOrCreateCarbonSession("hdfs://mast er:9000/opt/carbonstore") val src="hdfs://master:9000/data/carbondata" benchmark{carbon.sql(s"load data inpath '$src/order_detail.csv' into table pt.order_detail OPT IONS('DELIMITER'=',','fileheader'='id,order_code,sales_area_id,sales_id,order_inputer,pro_type, currency,exchange_rate,unit_cost_price,unit_selling_price,order_num,order_amount,order_disc ount,order_account_amount,order_time,delivery_channel,delivery_address,recipients,contact,d elivery_date,comments')") 导入方式二简介 CarbonData 支持将 DataFrame 转换成一个 Carbon File, 详情可参照文档 :Carbondata Interfaces 案例 // User can create a DataFrame from any data source or transformation. val df =... // Write data // User can write a DataFrame to a carbon file df.write.format("org.apache.spark.sql.carbonsource").option("tablename", "carbontable").mode(savemode.overwrite).save() 18 / 21

19 // read carbon data by data source API df = carboncontext.read.format("org.apache.spark.sql.carbonsource").option("tablename", "carbontable").load("/path") badrecord 处理简介 CarbonData Data Load 发生后有两种状态, 一种是成功, 另一种则是部分成功其中部分成功是由于数据导入过程发现了非法记录 ( 如一条记录的列数与表的列数不匹配, 官方称为 badrecord ),badrecord 存储在 carbon.badrecords.location 所配置的目录下案例案例如下 : CarbonProperties.getInstance().addProperty("carbon.badRecords.location","hdfs://master:900 0/data/carbondata/badrecords/") benchmark{carbon.sql(s"load data inpath '$src/order_detail.csv' into table pt.order_detail OPT IONS('DELIMITER'=',','bad_records_logger_enable'='true','fileheader'='id,order_code,sales_area_ id,sales_id,order_inputer,pro_type,currency,exchange_rate,unit_cost_price,unit_selling_price,or der_num,order_amount,order_discount,order_account_amount,order_time,delivery_channel,d elivery_address,recipients,contact,delivery_date,comments')") 数据校验简介在跑完 Load Data 作业后, 我们需要对数据进行多角度的校验, 如数据总量是否存在格式不匹配 ( 相应字段值为 NULL) 是否存在 badrecord 等案例 carobon.sql("select count(1) from pt.order_detail").show carbon.sql("select * from pt.order_detail").show 19 / 21

20 8.2 数据查询 CarbonData 支持两种方式的数据查询, 分别为 : - 通过 spark-shell - 通过 spark-sql api 查询方式一对于通过 spark-shell 的方式,CarbonData 查询的方式很简单, 只需在创建好 CarbonSession 的基础上执行如下语句 : carbon.sql("select * from pt.order_detail").show 同样的, 若要显示性能信息, 则添加 benchmark{: benchmark{carbon.sql("select * from pt.order_detail").show 查询方式二 CarbonData 还支持通过 spark-sql api 的方式对表进行查询, 如下所示 : carbondf.filter($"currency" === "1" and $"contact" === " ").count 同理, 若要显示性能信息, 则添加 benchmark{: benchmark{carbondf.filter($"currency" === "1" and $"contact" === " ").count 附 : 测试 SQL 脚本 OLAP 查询 20 / 21

21 Powered by TCPDF ( select recipients, sum(order_amount) sum_amount from pt.order_detail group by recipients h aving sum(order_amount) > 20 order by sum_amount desc limit 100 select delivery_date, avg(exchange_rate) from pt.order_detail group by delivery_date having av g(exchange_rate) > 100 limit 100 顺序查询 select * from pt.order_detail where order_discount < 0.5 and delivery_date > limit 10 select recipients, contact from pt.order_detail where order_num < 100 and order_amount < 随机查询 select * from pt.order_detail where contact= select * from pt.order_detail where sales_area_id=99 and sales_id= 结束语 CarbonData 作为大数据开源软件的 BLACKDUCK, 它的创新性和影响力已得到认可, 随着社区孵化进度的加速,CarbonData 在性能及通用性等方面势必有进一步可观的发展, 让我们一起拭目以待本博客文章除特别声明, 全部都是原创! 转载本文请加上 : 转载自过往记忆 ( 本文链接 : () 21 / 21

Apache CarbonData集群模式使用指南

Apache CarbonData集群模式使用指南我们在 Apache CarbonData 快速入门编程指南文章中介绍了如何快速使用 Apache CarbonData, 为了简单起见, 我们展示了如何在单机模式下使用 Apache CarbonData 但是生产环境下一般都是使用集群模式, 本文主要介绍如何在集群模式下使用 Apache CarbonData 启动 Spark shell 这里以 Spark shell 模式进行介绍,master