操作场景

本章节根据超过50个测试用例总结得出建议，帮助用户创建拥有更高查询性能的CarbonData表。

操作步骤

• 如果待创建的表有一个常用于过滤的列 ，例如80%以上的场景使用此列过滤。

  针对此类场景，调优方法如下：

  将常用于过滤的列放在sort_columns第一列。

  例如，msisdn作为过滤条件在查询中使用的最多，则将其放在第一列，因为msisdn的Cardinality为3千万，所以不需要做字典编码。创建表的命令如下，其中采用msisdn作为过滤条件的查询性能将会很好。

  create table carbondata_table(
      msisdn String,
      ...
      )STORED BY ‘org.apache.carbondata.format‘ TBLPROPERTIES (‘SORT_COLUMS‘=‘msisdn‘,‘DICTIONARY_EXCLUDE‘=‘msisdn,..‘,‘DICTIONARY_INCLUDE‘=‘...‘);

• 如果待创建的表有多个常用于过滤的列。

  针对此类场景，调优方法如下：

  为常用的过滤列创建索引。

  例如，如果msisdn，host和dime_1是过滤经常使用的列，根据cardinality，sort_columns列的顺序是dime_1-> host-> msisdn…；因为msisdn和host的Cardinality超过百万，所以不需要创建字典编码。创建表命令如下，以下命令可提高dime_1，host和msisdn上的过滤性能和压缩率。

  create table carbondata_table(
      dime_1 String,
      host String,
      msisdn String,
      dime_2 String,
      dime_3 String,
      ...
      )STORED BY ‘org.apache.carbondata.format‘ 
  TBLPROPERTIES (‘SORT_COLUMS‘=‘dime_1,host,msisdn‘,‘DICTIONARY_EXCLUDE‘=‘msisdn,host..‘,‘DICTIONARY_INCLUDE‘=‘dime_1..‘);

• 如果每个用于过滤的列的频率相当。

  针对此类场景，调优方法如下：

  sort_columns按照cardinality从低到高的顺序排列，并对高Cardinality列不做字典编码。

  创建表的命令如下：

  create table carbondata_table(
      Dime_1 String,
      BEGIN_TIME bigint,
      HOST String,
      MSISDN String,
      ...
      )STORED BY ‘org.apache.carbondata.format‘ 
  TBLPROPERTIES (‘SORT_COLUMS‘=‘dime_2,dime_3,dime_1, BEGIN_TIME,host,msisdn‘ ‘DICTIONARY_EXCLUDE‘=‘MSISDN,HOST,..‘,‘DICTIONARY_INCLUDE‘=‘Dime_1,BEGIN_TIME..‘);

• 按照维度的cardinality从低到高创建表后，再为高Cardinality列创建SECONDARY INDEX。创建索引的语句如下：

  create index carbondata_table_index_msidn on tablecarbondata_table (
  MSISDN String) as ‘org.apache.carbondata.format‘ TBLPROPERTIES (‘table_blocksize‘=‘128‘);
  create index carbondata_table_index_host on tablecarbondata_table (
  host String) as ‘org.apache.carbondata.format‘ TBLPROPERTIES (‘table_blocksize‘=‘128‘);

• 对于不需要高精度的度量，无需使用numeric (20,0)数据类型，建议使用double数据类型来替换numeric (20,0)数据类型，以提高查询性能。

  在一个测试用例中，使用double来替换numeric (20, 0)，查询时间从15秒降低到3秒，查询速度提高了5倍。创建表命令如下：

  create table carbondata_table(
      Dime_1 String,
      BEGIN_TIME bigint,
      HOST String,
      MSISDN String,
      counter_1 double,
      counter_2 double,
      ...
      counter_100 double,
      )STORED BY ‘org.apache.carbondata.format‘ 
      TBLPROPERTIES ( ‘DICTIONARY_EXCLUDE‘=‘MSISDN,HOST,IMSI‘,‘DICTIONARY_INCLUDE‘=‘Dime_1,END_TIME,BEGIN_TIME‘);

• 如果列值总是递增的，如start_time。

  例如，每天将数据加载到CarbonData，start_time是每次加载的增量。对于这种情况，建议将start_time列放在sort_columns的最后，因为总是递增的值可以始终使用最小/最大索引。创建表命令如下：

  create table carbondata_table(
      Dime_1 String,
      HOST String,
      MSISDN String,
      counter_1 double,
      counter_2 double,
      BEGIN_TIME bigint,
      ...
      counter_100 double,
      )STORED BY ‘org.apache.carbondata.format‘ 
      TBLPROPERTIES ( ‘SORT_COLUMS‘=‘dime_2,dime_3,dime_1..BEGIN_TIME‘,‘DICTIONARY_EXCLUDE‘=‘MSISDN,HOST,IMSI‘,‘DICTIONARY_INCLUDE‘=‘Dime_1,END_TIME,BEGIN_TIME‘);

  说明：

  对于维度是否需要生成字典，建议在没有足够的内存的情况下，不要把高cardinality列当作字典列。对高cardinality列生成字典会影响加载性能，需要更多的内存。CarbonData的默认行为是：CarbonData会对第一次数据加载的cardinality进行判断，如果cardinality小于100万，则生成字典。

   

  性能调优的相关配置

  操作场景

  CarbonData的性能与配置参数相关，本章节提供了能够提升性能的相关配置介绍。

  操作步骤

  用于CarbonData查询的配置介绍，详情请参见表1和表2。

  表1 Shuffle过程中，启动Task的个数

  参数	spark.sql.shuffle.partitions
  所属配置文件	spark-defaults.conf
  适用于	数据查询
  场景描述	Spark shuffle时启动的Task个数。
  如何调优	一般建议将该参数值设置为执行器核数的1到2倍。例如，在聚合场景中，将task个数从200减少到32，有些查询的性能可提升2倍。

  表2 设置用于CarbonData查询的Executor个数、CPU核数以及内存大小

  参数	SPARK_EXECUTOR_CORES SPARK_EXECUTOR_INSTANCES SPARK_EXECUTOR_MEMORY
  所属配置文件	spark-defaults.conf
  适用于	数据查询
  场景描述	设置用于CarbonData查询的Executor个数、CPU核数以及内存大小。
  如何调优	在银行方案中，为每个执行器提供4个CPU内核和15GB内存，可以获得良好的性能。这2个值并不意味着越多越好，在资源有限的情况下，需要正确配置。例如，在银行方案中，每个节点有足够的32个CPU核，而只有64GB的内存，这个内存是不够的。例如，当每个执行器有4个内核和12GB内存，有时在查询期间发生垃圾收集（GC），会导致查询时间从3秒增加到超过15秒。在这种情况下需要增加内存或减少CPU内核。

  用于CarbonData数据加载的配置参数，详情请参见表3、表4和表5。

  表3 设置数据加载使用的CPU core数量

  参数	carbon.number.of.cores.while.loading
  所属配置文件	carbon.properties
  适用于	数据加载
  场景描述	数据加载过程中，设置处理数据使用的CPU core数量。
  如何调优	如果有更多的CPU个数，那么可以增加CPU值来提高性能。例如，将该参数值从2增加到4，那么CSV文件读取性能可以增加大约1倍。

  表4 是否使用YARN本地目录进行多磁盘数据加载

  参数	carbon.use.local.dir
  所属配置文件	carbon.properties
  适用于	数据加载
  场景描述	是否使用YARN本地目录进行多磁盘数据加载。
  如何调优	如果将该参数值设置为“true”，CarbonData将使用YARN本地目录进行多表加载磁盘负载平衡，以提高数据加载性能。

  表5 加载时是否使用多路径

  参数	carbon.use.multiple.temp.dir
  所属配置文件	carbon.properties
  适用于	数据加载
  场景描述	是否使用多个临时目录存储sort临时文件。
  如何调优	设置为true，则数据加载时使用多个临时目录存储sort临时文件。此配置能提高数据加载性能并避免磁盘单点故障。

  用于CarbonData数据加载和数据查询的配置参数，详情请参见表6。

  表6 设置数据加载和查询使用的CPU core数量

  参数	carbon.compaction.level.threshold
  所属配置文件	carbon.properties
  适用于	数据加载和查询
  场景描述	对于minor压缩，在阶段1中要合并的segment数量和在阶段2中要合并的已压缩的segment数量。
  如何调优	每次CarbonData加载创建一个segment，如果每次加载的数据量较小，将在一段时间内生成许多小文件，影响查询性能。配置该参数将小的segment合并为一个大的segment，然后对数据进行排序，可提高查询性能。 压缩的策略根据实际的数据大小和可用资源决定。如某银行1天加载一次数据，且加载数据选择在晚上无查询时进行，有足够的资源，压缩策略可选择为6、5。

   

  beeline连接

   启动CarbonThriftServer服务

  #!/bin/bash
  nohup bin/spark-submit --master yarn --deploy-mode client --driver-memory 1g --num-executors 1 --executor-memory 1g --executor-cores 1 --conf spark.sql.shuffle.partitions=500 --conf spark.sql.auto.repartition=true --conf spark.shuffle.service.enabled=true --conf spark.executor.memoryOverhead=1g --conf spark.sql.autoBroadcastJoinThreshold=10485760 --class org.apache.carbondata.spark.thriftserver.CarbonThriftServer ../carbonlib/apache-carbondata-1.5.3-bin-spark2.3.2-hadoop2.7.2.jar >> ../logs/carbon_thrift.log 2>&1 &

  连接thriftserver

  ./beeline -u jdbc:hive2://ip:10000

  示例代码

  SparkStreaming数据入库至carbondata

  package com.taiji.streaming
  
  import java.io._
  import java.text.SimpleDateFormat
  import java.util.Properties
  import java.util.concurrent.ConcurrentHashMap
  
  import com.taiji.utils.ConfigurationXml
  import org.apache.commons.lang.StringUtils
  import org.apache.curator.framework.CuratorFrameworkFactory
  import org.apache.curator.retry.ExponentialBackoffRetry
  import org.apache.kafka.clients.consumer.ConsumerRecord
  import org.apache.kafka.common.TopicPartition
  import org.apache.kafka.common.serialization.StringDeserializer
  import org.apache.spark.rdd.RDD
  import org.apache.spark.sql.CarbonSession._
  import org.apache.spark.sql._
  import org.apache.spark.sql.types.{StringType, StructField, StructType}
  import org.apache.spark.streaming.dstream.InputDStream
  import org.apache.spark.streaming.kafka010._
  import org.apache.spark.streaming.{Seconds, StreamingContext, Time}
  import org.slf4j.LoggerFactory
  
  import scala.collection.mutable
  import scala.util.control.Breaks.{break, breakable}
  
  
  /**
    * sparkstreaming通过消费多个topic的数据进行分类存放。
    * 目前测试发现此种方式是效率最高的每分钟可消费1000w左右的数据
    *
    * spark 2.3.2
    * kafka服务端版本是2.10   spark-streaming-kafka-0-10_2.11-2.3.2
    *
    * 数据写入carbondata
    */
  object CoreRealtimeStreaming {
  
    private val log = LoggerFactory.getLogger("coreStreaming")
  
    def main(args: Array[String]): Unit = {
  
  
        val spark = SparkSession
          .builder()
          .appName("Streaming")
          .config("spark.streaming.kafka.maxRatePerPartition", kafkaMaxRatePerPar)
          .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          //      .config("spark.executor.memoryOverhead", 4096)
          //      .config("spark.yarn.executor.memoryOverhead", 4096)
          .enableHiveSupport()
          .getOrCreateCarbonSession()
  
  
        val ssc = new StreamingContext(spark.sparkContext, Seconds(time.toLong))
        spark.sql("use core")
        //创建carbon表
        createCarbonTable(spark)
        val sc = spark.sparkContext
        val schemaPathBro = sc.broadcast(schemaPath)
  
        val configurationMap = ConfigurationXml.testParserXml(xmlPath)
        val colAndPosMap = ConfigurationXml.getColAndPosMap
  
        for (tp <- 0 until tpArray.size) {
  
          val topic = tpArray(tp)
          val groupId = topic.split("_topic", -1)(0) + "_group"
          val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> topic_brokers,
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "auto.offset.reset" -> "latest",
            "group.id" -> groupId,
            "enable.auto.commit" -> "false"
          )
          log.info("consumer topic is: " + topic)
  
          val kafkaStream = createDirectKafkaStream(ssc, kafkaParams, topic, groupId)
  
          var offsetRanges = Array[OffsetRange]()
  
          val resourceRdd = kafkaStream.transform { rdd =>
            offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges //得到该 rdd 对应 kafka 的消息的 offset
            rdd
          }.map(rdd => {
            rdd.value()
          })
  
          resourceRdd.foreachRDD((rdd, time: Time) => {
            if (!rdd.isEmpty()) {
              //根据配置文件构造schema
              val properties = getSchemaProperties(schemaPathBro.value)
  
              val fields = properties.getProperty(topic)
              val schema = StructType(fields.split(",", -1).map(t => {
                StructField(t, StringType, true)
              }))
  
              // 代码的具体逻辑
              val correctBatchData = rdd.map(str => {
                val trksj = new SimpleDateFormat("yyyyMMdd").format(System.currentTimeMillis())
                val appendStr = str + "\t" + trksj
                val dataStr = appendStr.split("\t", -1)
                Row.fromSeq(dataStr)
              })
  
              val correctDf = spark.createDataFrame(correctBatchData, schema)
  
              val tableName = "bcp_" + topic.split("_topic", -1)(0)
              
              //默认存储位parquet
              //errorDataDf.write.mode(SaveMode.Append).partitionBy("TRKSJ").saveAsTable(tableName)
              //carbondata和partition不能同时使用，如下不支持分区
              //errorDataDf.write.mode(SaveMode.Append).format("carbondata").partitionBy("TRKSJ").saveAsTable(tableName)
              CarbonSparkStreamingFactory.getStreamSparkStreamingWriter(spark, "default", tableName)
                .mode(SaveMode.Append)
                .writeStreamData(correctDf, time)
  
              }
            }
            //记录偏移量
            storeOffsets(offsetRanges, groupId)
          })
        }
        ssc.start()
        ssc.awaitTermination()
        spark.stop()
      } catch {
        case e: Exception => {
          log.error("CoreRealtimeStreaming is error. ", e)
        }
      }
    }
  
    def closeZkClient() {
      if (client != null) {
        client.close()
      }
    }
  
    val config = {
      val configFile = new File("config.xml")
      if (!configFile.exists || !configFile.isFile) System.exit(-1)
      val configInfo = new Properties
  
      try
        configInfo.loadFromXML(new FileInputStream(configFile))
      catch {
        case e: IOException =>
          log.error("config.xml Incorrect format!")
          System.exit(-1)
      }
      configInfo
    }
  
    val client = {
  
      val zkHost = config.getProperty("zk_host")
      if (zkHost == null || zkHost.length == 0) {
        log.error("zkHost is not config.")
        System.exit(-1)
      }
      val client = CuratorFrameworkFactory
        .builder
        .connectString(zkHost)
        .retryPolicy(new ExponentialBackoffRetry(1000, 3))
        .namespace("kafka")
        .build()
      client.start()
      client
    }
  
    // offset 路径起始位置
    val Globe_kafkaOffsetPath = "/consumers/streaming"
  
    // 路径确认函数  确认ZK中路径存在，不存在则创建该路径
    def checkZKPathExists(path: String) = {
  
      if (client.checkExists().forPath(path) == null) {
        client.create().creatingParentsIfNeeded().forPath(path)
      }
  
    }
  
    // 保存 新的 offset
    def storeOffsets(offsetRange: Array[OffsetRange], groupName: String) = {
  
      for (o <- offsetRange) {
        val zkPath = s"${Globe_kafkaOffsetPath}/${groupName}/${o.topic}/${o.partition}"
  
        // 向对应分区第一次写入或者更新Offset 信息
        log.info("Topic：" + o.topic + ", Partition:" + o.partition + ", Offset:" + o.untilOffset + " to zk")
        if (client.checkExists().forPath(zkPath) == null) {
          client.create().creatingParentsIfNeeded().forPath(zkPath)
        }
        client.setData().forPath(zkPath, o.untilOffset.toString.getBytes())
      }
    }
  
    def getFromOffset(topic: String, groupName: String): (Map[TopicPartition, Long], Int) = {
  
      // Kafka 0.8和0.10的版本差别，0.10 为 TopicPartition   0.8 TopicAndPartition
  
      // 读取ZK中保存的Offset，作为Dstrem的起始位置。如果没有则创建该路径，并从 0 开始Dstream
      val zkTopicPath = s"${Globe_kafkaOffsetPath}/${groupName}/${topic}"
  
      // 检查路径是否存在
      checkZKPathExists(zkTopicPath)
  
      // 获取topic的子节点，即 分区
      val childrens = client.getChildren().forPath(zkTopicPath)
  
      import scala.collection.JavaConversions._
      // 遍历分区
      val offSets: mutable.Buffer[(TopicPartition, Long)] = for {
        p <- childrens
      } yield {
        // 遍历读取子节点中的数据：即 offset
        val offsetData = client.getData().forPath(s"$zkTopicPath/$p")
        // 将offset转为Long
        val offSet = java.lang.Long.valueOf(new String(offsetData)).toLong
        // 返回  (TopicPartition, Long)
        (new TopicPartition(topic, Integer.parseInt(p)), offSet)
      }
      println(offSets.toMap)
      if (offSets.isEmpty) {
        (offSets.toMap, 0)
      } else {
        (offSets.toMap, 1)
      }
    }
  
    def createDirectKafkaStream(ssc: StreamingContext, kafkaParams: Map[String, Object], topic: String,
                                groupName: String): InputDStream[ConsumerRecord[String, String]] = {
  
      // get offset  flag = 1  表示基于已有的offset计算  flag = 表示从头开始(最早或者最新，根据Kafka配置)
      val (fromOffsets, flag) = getFromOffset(topic, groupName)
      var kafkaStream: InputDStream[ConsumerRecord[String, String]] = null
      if (flag == 1) {
        // 加上消息头
        //val messageHandler = (mmd:MessageAndMetadata[String, String]) => (mmd.topic, mmd.message())
        println(fromOffsets)
        kafkaStream = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe(Array(topic), kafkaParams, fromOffsets))
        println(fromOffsets)
  
      } else {
        kafkaStream = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe(Array(topic), kafkaParams))
      }
      kafkaStream
    }
  
    def getSchemaProperties(path: String): Properties = {
  
      //保证读取出来的sql顺序与文件顺序一致
      val properties = new Properties()
  
      // 使用InPutStream流读取properties文件
      val bufferedReader = new BufferedReader(new FileReader(path))
      properties.load(bufferedReader)
  
      properties
    }
  
    /**
      * create carbondata table
      *
      * @param spark   sparksession
      */
    def createCarbonTable(spark: SparkSession): Unit = {
      
      val objTags = "bcp_obj_tags"
      spark.sql(
        s"""
           | CREATE TABLE IF NOT EXISTS $objTags(
           | obj_lx String,
           | obj_val String,
           | tags String,
           | CJSJ int,
           | TRKSJ int)
           | STORED BY ‘carbondata‘
           | TBLPROPERTIES(
           | ‘streaming‘=‘true‘,
           | ‘sort_columns‘=‘TRKSJ‘)
           | """.stripMargin)
    }
  }