9642

Build # 9642

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/f4188ccdcf5c5d0003a0076200cea71cfdc1449a">f4188ccdc<a href="https://github.com/apache/carbondata/commit/f4188ccdcf5c5d0003a0076200cea71cfdc1449a">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/97b39d83fbaf1ac26d0f1fc234da75cfdb9e2b51">97b39d83f</a>">97b39d83f</a><a href="https://github.com/apache/carbondata/commit/f4188ccdcf5c5d0003a0076200cea71cfdc1449a"> into c94c8ce56">c94c8ce56</a>

Run Details

67 of 67 new or added lines in 6 files covered. (100.0%)

61329 of 77012 relevant lines covered (79.64%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.89

/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.carbondata.spark.load

import java.text.SimpleDateFormat
import java.util.{Date, Locale}

import scala.collection.mutable.ArrayBuffer

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.mapred.JobConf
import org.apache.hadoop.mapreduce.{TaskAttemptID, TaskType}
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
import org.apache.hadoop.mapreduce.task.{JobContextImpl, TaskAttemptContextImpl}
import org.apache.spark.Partition
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
import org.apache.spark.sql.util.SparkSQLUtil
import org.apache.spark.sql.util.SparkSQLUtil.sessionState

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.util.ThreadLocalSessionInfo
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.spark.util.CommonUtil

object CsvRDDHelper {

  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  def csvGetPartition(
      spark: SparkSession,
      model: CarbonLoadModel,
      hadoopConf: Configuration): Array[FilePartition] = {
    val defaultMaxSplitBytes = sessionState(spark).conf.filesMaxPartitionBytes
    val openCostInBytes = sessionState(spark).conf.filesOpenCostInBytes
    val defaultParallelism = spark.sparkContext.defaultParallelism
    CommonUtil.configureCSVInputFormat(hadoopConf, model)
    hadoopConf.set(FileInputFormat.INPUT_DIR, model.getFactFilePath)
    val jobConf = new JobConf(hadoopConf)
    SparkHadoopUtil.get.addCredentials(jobConf)
    val jobContext = new JobContextImpl(jobConf, null)
    val inputFormat = new CSVInputFormat()
    val rawSplits = inputFormat.getSplits(jobContext).toArray
    val splitFiles = rawSplits.map { split =>
      val fileSplit = split.asInstanceOf[FileSplit]
      PartitionedFile(
        InternalRow.empty,
        fileSplit.getPath.toString,
        fileSplit.getStart,
        fileSplit.getLength,
        fileSplit.getLocations)
    }.sortBy(_.length)(implicitly[Ordering[Long]].reverse)
    val totalBytes = splitFiles.map(_.length + openCostInBytes).sum
    val bytesPerCore = totalBytes / defaultParallelism

    val maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
    LOGGER.info(s"Planning scan with bin packing, max size: $maxSplitBytes bytes, " +
                s"open cost is considered as scanning $openCostInBytes bytes.")

    val partitions = new ArrayBuffer[FilePartition]
    val currentFiles = new ArrayBuffer[PartitionedFile]
    var currentSize = 0L

    def closePartition(): Unit = {
      if (currentFiles.nonEmpty) {
        val newPartition =
          FilePartition(
            partitions.size,
            currentFiles.toArray.toSeq)
        partitions += newPartition
      }
      currentFiles.clear()
      currentSize = 0
    }

    splitFiles.foreach { file =>
      if (currentSize + file.length > maxSplitBytes) {
        closePartition()
      }
      // Add the given file to the current partition.
      currentSize += file.length + openCostInBytes
      currentFiles += file
    }
    closePartition()
    partitions.toArray
  }

  /**
   * createsw a RDD that does reading of multiple CSV files
   */
  def csvFileScanRDD(
      spark: SparkSession,
      model: CarbonLoadModel,
      hadoopConf: Configuration
  ): RDD[InternalRow] = {
    // 1. partition
    val partitions = csvGetPartition(spark, model, hadoopConf)

    // 2. read function
    val serializableConfiguration = SparkSQLUtil.getSerializableConfigurableInstance(hadoopConf)
    val readFunction = new (PartitionedFile => Iterator[InternalRow]) with Serializable {
      override def apply(file: PartitionedFile): Iterator[InternalRow] = {
        new Iterator[InternalRow] {
          val jobTrackerId: String = {
            val formatter = new SimpleDateFormat("yyyyMMddHHmmss", Locale.US)
            formatter.format(new Date())
          }
          ThreadLocalSessionInfo.setConfigurationToCurrentThread(serializableConfiguration.value)
          val attemptId = new TaskAttemptID(jobTrackerId, 0, TaskType.MAP, 0, 0)
          val hadoopAttemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration,
            attemptId)
          val inputSplit =
            new FileSplit(new Path(file.filePath), file.start, file.length, file.locations)
          var finished = false
          val inputFormat = new CSVInputFormat()
          val reader = inputFormat.createRecordReader(inputSplit, hadoopAttemptContext)
          reader.initialize(inputSplit, hadoopAttemptContext)

          override def hasNext: Boolean = {
            if (!finished) {
              if (reader != null) {
                if (reader.nextKeyValue()) {
                  true
                } else {
                  finished = true
                  reader.close()
                  false
                }
              } else {
                finished = true
                false
              }
            } else {
              false
            }
          }

          override def next(): InternalRow = {
            new GenericInternalRow(reader.getCurrentValue.get().asInstanceOf[Array[Any]])
          }
        }
      }
    }
    new FileScanRDD(spark, readFunction, partitions)
  }

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.spark.load
19
20	import java.text.SimpleDateFormat
21	import java.util.{Date, Locale}
22
23	import scala.collection.mutable.ArrayBuffer
24
25	import org.apache.hadoop.conf.Configuration
26	import org.apache.hadoop.fs.Path
27	import org.apache.hadoop.mapred.JobConf
28	import org.apache.hadoop.mapreduce.{TaskAttemptID, TaskType}
29	import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
30	import org.apache.hadoop.mapreduce.task.{JobContextImpl, TaskAttemptContextImpl}
31	import org.apache.spark.Partition
32	import org.apache.spark.deploy.SparkHadoopUtil
33	import org.apache.spark.rdd.RDD
34	import org.apache.spark.sql.SparkSession
35	import org.apache.spark.sql.catalyst.InternalRow
36	import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
37	import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionedFile}
38	import org.apache.spark.sql.util.SparkSQLUtil
39	import org.apache.spark.sql.util.SparkSQLUtil.sessionState
40
41	import org.apache.carbondata.common.logging.LogServiceFactory
42	import org.apache.carbondata.core.datastore.impl.FileFactory
43	import org.apache.carbondata.core.util.ThreadLocalSessionInfo
44	import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
45	import org.apache.carbondata.processing.loading.model.CarbonLoadModel
46	import org.apache.carbondata.spark.util.CommonUtil
47
48	object CsvRDDHelper {
49
50	private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	1✔
51
52	def csvGetPartition(
53	spark: SparkSession,
54	model: CarbonLoadModel,
55	hadoopConf: Configuration): Array[FilePartition] = {
56	val defaultMaxSplitBytes = sessionState(spark).conf.filesMaxPartitionBytes	1✔
57	val openCostInBytes = sessionState(spark).conf.filesOpenCostInBytes	1✔
58	val defaultParallelism = spark.sparkContext.defaultParallelism	1✔
59	CommonUtil.configureCSVInputFormat(hadoopConf, model)	1✔
60	hadoopConf.set(FileInputFormat.INPUT_DIR, model.getFactFilePath)	1✔
61	val jobConf = new JobConf(hadoopConf)	1✔
62	SparkHadoopUtil.get.addCredentials(jobConf)	1✔
63	val jobContext = new JobContextImpl(jobConf, null)	1✔
64	val inputFormat = new CSVInputFormat()	1✔
65	val rawSplits = inputFormat.getSplits(jobContext).toArray	1✔
66	val splitFiles = rawSplits.map { split =>	1✔
67	val fileSplit = split.asInstanceOf[FileSplit]	1✔
68	PartitionedFile(	1✔
69	InternalRow.empty,	1✔
70	fileSplit.getPath.toString,	1✔
71	fileSplit.getStart,	1✔
72	fileSplit.getLength,	1✔
73	fileSplit.getLocations)	1✔
74	}.sortBy(_.length)(implicitly[Ordering[Long]].reverse)	1✔
75	val totalBytes = splitFiles.map(_.length + openCostInBytes).sum	1✔
76	val bytesPerCore = totalBytes / defaultParallelism	1✔
77
78	val maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))	1✔
79	LOGGER.info(s"Planning scan with bin packing, max size: $maxSplitBytes bytes, " +	1✔
80	s"open cost is considered as scanning $openCostInBytes bytes.")	1✔
81
82	val partitions = new ArrayBuffer[FilePartition]	1✔
83	val currentFiles = new ArrayBuffer[PartitionedFile]	1✔
84	var currentSize = 0L	1✔
85
86	def closePartition(): Unit = {
87	if (currentFiles.nonEmpty) {	1✔
88	val newPartition =	1✔
89	FilePartition(	1✔
90	partitions.size,	1✔
91	currentFiles.toArray.toSeq)	1✔
92	partitions += newPartition	1✔
93	}
94	currentFiles.clear()	1✔
95	currentSize = 0	1✔
96	}
97
98	splitFiles.foreach { file =>	1✔
99	if (currentSize + file.length > maxSplitBytes) {	1✔
100	closePartition()	1✔
101	}
102	// Add the given file to the current partition.
103	currentSize += file.length + openCostInBytes	1✔
104	currentFiles += file	1✔
105	}
106	closePartition()	1✔
107	partitions.toArray	1✔
108	}
109
110	/**
111	* createsw a RDD that does reading of multiple CSV files
112	*/
113	def csvFileScanRDD(
114	spark: SparkSession,
115	model: CarbonLoadModel,
116	hadoopConf: Configuration
117	): RDD[InternalRow] = {
118	// 1. partition
119	val partitions = csvGetPartition(spark, model, hadoopConf)	1✔
120
121	// 2. read function
122	val serializableConfiguration = SparkSQLUtil.getSerializableConfigurableInstance(hadoopConf)	1✔
123	val readFunction = new (PartitionedFile => Iterator[InternalRow]) with Serializable {	1✔
124	override def apply(file: PartitionedFile): Iterator[InternalRow] = {
125	new Iterator[InternalRow] {	1✔
126	val jobTrackerId: String = {	1✔
127	val formatter = new SimpleDateFormat("yyyyMMddHHmmss", Locale.US)	1✔
128	formatter.format(new Date())	1✔
129	}
130	ThreadLocalSessionInfo.setConfigurationToCurrentThread(serializableConfiguration.value)	1✔
131	val attemptId = new TaskAttemptID(jobTrackerId, 0, TaskType.MAP, 0, 0)	1✔
132	val hadoopAttemptContext = new TaskAttemptContextImpl(FileFactory.getConfiguration,	1✔
133	attemptId)	1✔
134	val inputSplit =	1✔
135	new FileSplit(new Path(file.filePath), file.start, file.length, file.locations)	1✔
136	var finished = false	1✔
137	val inputFormat = new CSVInputFormat()	1✔
138	val reader = inputFormat.createRecordReader(inputSplit, hadoopAttemptContext)	1✔
139	reader.initialize(inputSplit, hadoopAttemptContext)	1✔
140
141	override def hasNext: Boolean = {
142	if (!finished) {	1✔
143	if (reader != null) {	1✔
144	if (reader.nextKeyValue()) {	1✔
145	true	1✔
146	} else {
147	finished = true	1✔
148	reader.close()	1✔
149	false	1✔
150	}
151	} else {
152	finished = true	×
153	false	×
154	}
155	} else {
156	false	×
157	}
158	}
159
160	override def next(): InternalRow = {
161	new GenericInternalRow(reader.getCurrentValue.get().asInstanceOf[Array[Any]])	1✔
162	}
163	}
164	}
165	}
166	new FileScanRDD(spark, readFunction, partitions)	1✔
167	}
168
169	}	1✔

apache / carbondata / 9642

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous