You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/11/21 03:57:03 UTC
carbondata git commit: [CARBONDATA-1780] Create configuration from
SparkSession for data loading
Repository: carbondata
Updated Branches:
refs/heads/master 075809a81 -> cb6c27173
[CARBONDATA-1780] Create configuration from SparkSession for data loading
Create configuration from SparkSession for data loading, so that we can set configuration into SparkSession during dataloading.
This closes #1539
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/cb6c2717
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/cb6c2717
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/cb6c2717
Branch: refs/heads/master
Commit: cb6c2717347a1f98f48afdfba2967afe21960c0a
Parents: 075809a
Author: QiangCai <qi...@qq.com>
Authored: Mon Nov 20 18:38:20 2017 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Tue Nov 21 11:56:50 2017 +0800
----------------------------------------------------------------------
.../filesystem/AbstractDFSCarbonFile.java | 26 +++++++++----
.../datastore/filesystem/HDFSCarbonFile.java | 13 ++++++-
.../core/datastore/impl/FileFactory.java | 27 +++++++++++++-
.../apache/carbondata/core/util/CarbonUtil.java | 19 ++++++++++
.../load/DataLoadProcessBuilderOnSpark.scala | 10 ++---
.../spark/rdd/NewCarbonDataLoadRDD.scala | 5 ++-
.../carbondata/spark/util/CommonUtil.scala | 10 +++--
.../carbondata/spark/util/DataLoadingUtil.scala | 7 +++-
.../spark/util/GlobalDictionaryUtil.scala | 13 ++++---
.../scala/org/apache/spark/util/FileUtils.scala | 17 +++++++--
.../spark/rdd/CarbonDataRDDFactory.scala | 39 +++++++++++---------
.../org/apache/spark/sql/CarbonSource.scala | 1 +
.../command/management/LoadTableCommand.scala | 22 ++++++++---
.../spark/util/AllDictionaryTestCase.scala | 15 +++++---
.../util/ExternalColumnDictionaryTestCase.scala | 31 +++++++++++-----
.../streaming/StreamSinkFactory.scala | 8 +++-
16 files changed, 189 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
index 8a86d92..836143c 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
@@ -27,6 +27,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.util.CarbonUtil;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -38,13 +39,19 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
private static final LogService LOGGER =
LogServiceFactory.getLogService(AbstractDFSCarbonFile.class.getName());
protected FileStatus fileStatus;
+ protected Configuration hadoopConf;
public AbstractDFSCarbonFile(String filePath) {
+ this(filePath, FileFactory.getConfiguration());
+ }
+
+ public AbstractDFSCarbonFile(String filePath, Configuration hadoopConf) {
+ this.hadoopConf = hadoopConf;
filePath = filePath.replace("\\", "/");
Path path = new Path(filePath);
FileSystem fs;
try {
- fs = path.getFileSystem(FileFactory.getConfiguration());
+ fs = path.getFileSystem(this.hadoopConf);
fileStatus = fs.getFileStatus(path);
} catch (IOException e) {
LOGGER.error("Exception occurred:" + e.getMessage());
@@ -52,9 +59,14 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
}
public AbstractDFSCarbonFile(Path path) {
+ this(path, FileFactory.getConfiguration());
+ }
+
+ public AbstractDFSCarbonFile(Path path, Configuration hadoopConf) {
+ this.hadoopConf = hadoopConf;
FileSystem fs;
try {
- fs = path.getFileSystem(FileFactory.getConfiguration());
+ fs = path.getFileSystem(this.hadoopConf);
fileStatus = fs.getFileStatus(path);
} catch (IOException e) {
LOGGER.error("Exception occurred:" + e.getMessage());
@@ -69,7 +81,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
Path path = fileStatus.getPath();
FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
return fs.createNewFile(path);
} catch (IOException e) {
return false;
@@ -92,7 +104,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
FileSystem fs;
try {
if (null != fileStatus) {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
return fs.exists(fileStatus.getPath());
}
} catch (IOException e) {
@@ -116,7 +128,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
public boolean renameTo(String changetoName) {
FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
return fs.rename(fileStatus.getPath(), new Path(changetoName));
} catch (IOException e) {
LOGGER.error("Exception occurred:" + e.getMessage());
@@ -127,7 +139,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
public boolean delete() {
FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
return fs.delete(fileStatus.getPath(), true);
} catch (IOException e) {
LOGGER.error("Exception occurred:" + e.getMessage());
@@ -142,7 +154,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
@Override public boolean setLastModifiedTime(long timestamp) {
FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
fs.setTimes(fileStatus.getPath(), timestamp, timestamp);
} catch (IOException e) {
return false;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
index 19bdc7b..a2fa6b6 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
@@ -25,6 +25,7 @@ import org.apache.carbondata.common.logging.LogService;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -41,10 +42,18 @@ public class HDFSCarbonFile extends AbstractDFSCarbonFile {
super(filePath);
}
+ public HDFSCarbonFile(String filePath, Configuration hadoopConf) {
+ super(filePath, hadoopConf);
+ }
+
public HDFSCarbonFile(Path path) {
super(path);
}
+ public HDFSCarbonFile(Path path, Configuration hadoopConf) {
+ super(path, hadoopConf);
+ }
+
public HDFSCarbonFile(FileStatus fileStatus) {
super(fileStatus);
}
@@ -103,14 +112,14 @@ public class HDFSCarbonFile extends AbstractDFSCarbonFile {
@Override
public CarbonFile getParentFile() {
Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new HDFSCarbonFile(parent);
+ return null == parent ? null : new HDFSCarbonFile(parent, hadoopConf);
}
@Override
public boolean renameForce(String changetoName) {
FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
+ fs = fileStatus.getPath().getFileSystem(hadoopConf);
if (fs instanceof DistributedFileSystem) {
((DistributedFileSystem) fs).rename(fileStatus.getPath(), new Path(changetoName),
org.apache.hadoop.fs.Options.Rename.OVERWRITE);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
index 988f1e5..5e65914 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
@@ -124,6 +124,23 @@ public final class FileFactory {
}
}
+ public static CarbonFile getCarbonFile(String path, FileType fileType,
+ Configuration hadoopConf) {
+ switch (fileType) {
+ case LOCAL:
+ return new LocalCarbonFile(getUpdatedFilePath(path, fileType));
+ case HDFS:
+ case S3:
+ return new HDFSCarbonFile(path, hadoopConf);
+ case ALLUXIO:
+ return new AlluxioCarbonFile(path);
+ case VIEWFS:
+ return new ViewFSCarbonFile(path);
+ default:
+ return new LocalCarbonFile(getUpdatedFilePath(path, fileType));
+ }
+ }
+
public static DataInputStream getDataInputStream(String path, FileType fileType)
throws IOException {
return getDataInputStream(path, fileType, -1);
@@ -131,6 +148,12 @@ public final class FileFactory {
public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize)
throws IOException {
+ return getDataInputStream(path, fileType, bufferSize, configuration);
+ }
+
+ public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
+ Configuration hadoopConf)
+ throws IOException {
path = path.replace("\\", "/");
boolean gzip = path.endsWith(".gz");
boolean bzip2 = path.endsWith(".bz2");
@@ -151,7 +174,7 @@ public final class FileFactory {
case VIEWFS:
case S3:
Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
+ FileSystem fs = pt.getFileSystem(hadoopConf);
if (bufferSize == -1) {
stream = fs.open(pt);
} else {
@@ -164,7 +187,7 @@ public final class FileFactory {
codecName = BZip2Codec.class.getName();
}
if (null != codecName) {
- CompressionCodecFactory ccf = new CompressionCodecFactory(configuration);
+ CompressionCodecFactory ccf = new CompressionCodecFactory(hadoopConf);
CompressionCodec codec = ccf.getCodecByClassName(codecName);
stream = codec.createInputStream(stream);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index 9d6acb6..02fe054 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -1358,6 +1358,25 @@ public final class CarbonUtil {
return readLine;
}
+ public static String readHeader(String csvFilePath,
+ Configuration hadoopConf) throws IOException {
+
+ DataInputStream fileReader = null;
+ BufferedReader bufferedReader = null;
+ String readLine = null;
+
+ try {
+ fileReader = FileFactory.getDataInputStream(
+ csvFilePath, FileFactory.getFileType(csvFilePath), -1, hadoopConf);
+ bufferedReader = new BufferedReader(new InputStreamReader(fileReader,
+ Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
+ readLine = bufferedReader.readLine();
+ } finally {
+ CarbonUtil.closeStreams(fileReader, bufferedReader);
+ }
+ return readLine;
+ }
+
/**
* Below method will create string like "***********"
*
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
index ca2ca2e..776973b 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessBuilderOnSpark.scala
@@ -49,21 +49,21 @@ object DataLoadProcessBuilderOnSpark {
def loadDataUsingGlobalSort(
sc: SparkContext,
dataFrame: Option[DataFrame],
- model: CarbonLoadModel): Array[(String, (LoadMetadataDetails, ExecutionErrors))] = {
+ model: CarbonLoadModel,
+ hadoopConf: Configuration): Array[(String, (LoadMetadataDetails, ExecutionErrors))] = {
val originRDD = if (dataFrame.isDefined) {
dataFrame.get.rdd
} else {
// input data from files
- val hadoopConfiguration = new Configuration()
- CommonUtil.configureCSVInputFormat(hadoopConfiguration, model)
- hadoopConfiguration.set(FileInputFormat.INPUT_DIR, model.getFactFilePath)
+ CommonUtil.configureCSVInputFormat(hadoopConf, model)
+ hadoopConf.set(FileInputFormat.INPUT_DIR, model.getFactFilePath)
val columnCount = model.getCsvHeaderColumns.length
new NewHadoopRDD[NullWritable, StringArrayWritable](
sc,
classOf[CSVInputFormat],
classOf[NullWritable],
classOf[StringArrayWritable],
- hadoopConfiguration)
+ hadoopConf)
.map(x => DataLoadProcessorStepOnSpark.toStringArrayRow(x._2, columnCount))
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
index 0fed5a7..f948ac8 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala
@@ -179,7 +179,8 @@ class NewCarbonDataLoadRDD[K, V](
sc: SparkContext,
result: DataLoadResult[K, V],
carbonLoadModel: CarbonLoadModel,
- blocksGroupBy: Array[(String, Array[BlockDetails])])
+ blocksGroupBy: Array[(String, Array[BlockDetails])],
+ @transient hadoopConf: Configuration)
extends CarbonRDD[(K, V)](sc, Nil) {
sc.setLocalProperty("spark.scheduler.pool", "DDL")
@@ -192,7 +193,7 @@ class NewCarbonDataLoadRDD[K, V](
private val confBytes = {
val bao = new ByteArrayOutputStream()
val oos = new ObjectOutputStream(bao)
- sc.hadoopConfiguration.write(oos)
+ hadoopConf.write(oos)
oos.close()
CompressorFactory.getInstance().getCompressor.compressByte(bao.toByteArray)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index c529cd3..838ab89 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -607,7 +607,9 @@ object CommonUtil {
}
}
- def getCsvHeaderColumns(carbonLoadModel: CarbonLoadModel): Array[String] = {
+ def getCsvHeaderColumns(
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration): Array[String] = {
val delimiter = if (StringUtils.isEmpty(carbonLoadModel.getCsvDelimiter)) {
CarbonCommonConstants.COMMA
} else {
@@ -618,7 +620,7 @@ object CommonUtil {
val csvColumns = if (StringUtils.isBlank(csvHeader)) {
// read header from csv file
csvFile = carbonLoadModel.getFactFilePath.split(",")(0)
- csvHeader = CarbonUtil.readHeader(csvFile)
+ csvHeader = CarbonUtil.readHeader(csvFile, hadoopConf)
if (StringUtils.isBlank(csvHeader)) {
throw new CarbonDataLoadingException("First line of the csv is not valid.")
}
@@ -638,10 +640,10 @@ object CommonUtil {
} else {
LOGGER.error(
"CSV header in input file is not proper. Column names in schema and csv header are not "
- + "the same. Input file : " + csvFile)
+ + "the same. Input file : " + CarbonUtil.removeAKSK(csvFile))
throw new CarbonDataLoadingException(
"CSV header in input file is not proper. Column names in schema and csv header are not "
- + "the same. Input file : " + csvFile)
+ + "the same. Input file : " + CarbonUtil.removeAKSK(csvFile))
}
}
csvColumns
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
index f359377..77f75f2 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala
@@ -22,6 +22,7 @@ import scala.collection.JavaConverters._
import scala.collection.mutable
import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql.util.CarbonException
import org.apache.carbondata.common.constants.LoggerAction
@@ -195,7 +196,8 @@ object DataLoadingUtil {
carbonProperty: CarbonProperties,
options: immutable.Map[String, String],
optionsFinal: mutable.Map[String, String],
- carbonLoadModel: CarbonLoadModel): Unit = {
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration): Unit = {
carbonLoadModel.setTableName(table.getTableName)
carbonLoadModel.setDatabaseName(table.getDatabaseName)
carbonLoadModel.setTablePath(table.getTablePath)
@@ -306,7 +308,8 @@ object DataLoadingUtil {
carbonLoadModel.setCsvDelimiter(CarbonUtil.unescapeChar(delimeter))
carbonLoadModel.setCsvHeader(fileHeader)
carbonLoadModel.setColDictFilePath(column_dict)
- carbonLoadModel.setCsvHeaderColumns(CommonUtil.getCsvHeaderColumns(carbonLoadModel))
+ carbonLoadModel.setCsvHeaderColumns(
+ CommonUtil.getCsvHeaderColumns(carbonLoadModel, hadoopConf))
val validatedMaxColumns = CommonUtil.validateMaxColumns(
carbonLoadModel.getCsvHeaderColumns,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index 0bf2b16..c931c44 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -352,10 +352,10 @@ object GlobalDictionaryUtil {
* @param carbonLoadModel carbon data load model
*/
def loadDataFrame(sqlContext: SQLContext,
- carbonLoadModel: CarbonLoadModel): DataFrame = {
- val hadoopConfiguration = new Configuration()
- CommonUtil.configureCSVInputFormat(hadoopConfiguration, carbonLoadModel)
- hadoopConfiguration.set(FileInputFormat.INPUT_DIR, carbonLoadModel.getFactFilePath)
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration): DataFrame = {
+ CommonUtil.configureCSVInputFormat(hadoopConf, carbonLoadModel)
+ hadoopConf.set(FileInputFormat.INPUT_DIR, carbonLoadModel.getFactFilePath)
val columnNames = carbonLoadModel.getCsvHeaderColumns
val schema = StructType(columnNames.map[StructField, Array[StructField]] { column =>
StructField(column, StringType)
@@ -367,7 +367,7 @@ object GlobalDictionaryUtil {
classOf[CSVInputFormat],
classOf[NullWritable],
classOf[StringArrayWritable],
- hadoopConfiguration).setName("global dictionary").map[Row] { currentRow =>
+ hadoopConf).setName("global dictionary").map[Row] { currentRow =>
row.setValues(currentRow._2.get())
}
sqlContext.createDataFrame(rdd, schema)
@@ -677,6 +677,7 @@ object GlobalDictionaryUtil {
def generateGlobalDictionary(sqlContext: SQLContext,
carbonLoadModel: CarbonLoadModel,
tablePath: String,
+ hadoopConf: Configuration,
dataFrame: Option[DataFrame] = None): Unit = {
try {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
@@ -694,7 +695,7 @@ object GlobalDictionaryUtil {
if (StringUtils.isEmpty(allDictionaryPath)) {
LOGGER.info("Generate global dictionary from source data files!")
// load data by using dataSource com.databricks.spark.csv
- var df = dataFrame.getOrElse(loadDataFrame(sqlContext, carbonLoadModel))
+ var df = dataFrame.getOrElse(loadDataFrame(sqlContext, carbonLoadModel, hadoopConf))
var headers = carbonLoadModel.getCsvHeaderColumns
headers = headers.map(headerName => headerName.trim)
val colDictFilePath = carbonLoadModel.getColDictFilePath
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
index 46311db..fcd460e 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
@@ -17,6 +17,8 @@
package org.apache.spark.util
+import org.apache.hadoop.conf.Configuration
+
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.filesystem.CarbonFile
@@ -28,11 +30,14 @@ object FileUtils {
/**
* append all csv file path to a String, file path separated by comma
*/
- private def getPathsFromCarbonFile(carbonFile: CarbonFile, stringBuild: StringBuilder): Unit = {
+ private def getPathsFromCarbonFile(
+ carbonFile: CarbonFile,
+ stringBuild: StringBuilder,
+ hadoopConf: Configuration): Unit = {
if (carbonFile.isDirectory) {
val files = carbonFile.listFiles()
for (j <- 0 until files.size) {
- getPathsFromCarbonFile(files(j), stringBuild)
+ getPathsFromCarbonFile(files(j), stringBuild, hadoopConf)
}
} else {
val path = carbonFile.getAbsolutePath
@@ -55,6 +60,10 @@ object FileUtils {
*
*/
def getPaths(inputPath: String): String = {
+ getPaths(inputPath, FileFactory.getConfiguration)
+ }
+
+ def getPaths(inputPath: String, hadoopConf: Configuration): String = {
if (inputPath == null || inputPath.isEmpty) {
throw new DataLoadingException("Input file path cannot be empty.")
} else {
@@ -62,12 +71,12 @@ object FileUtils {
val filePaths = inputPath.split(",")
for (i <- 0 until filePaths.size) {
val fileType = FileFactory.getFileType(filePaths(i))
- val carbonFile = FileFactory.getCarbonFile(filePaths(i), fileType)
+ val carbonFile = FileFactory.getCarbonFile(filePaths(i), fileType, hadoopConf)
if (!carbonFile.exists()) {
throw new DataLoadingException(
s"The input file does not exist: ${CarbonUtil.removeAKSK(filePaths(i))}" )
}
- getPathsFromCarbonFile(carbonFile, stringBuild)
+ getPathsFromCarbonFile(carbonFile, stringBuild, hadoopConf)
}
if (stringBuild.nonEmpty) {
stringBuild.substring(0, stringBuild.size - 1)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 47e016b..95ccc92 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -268,6 +268,7 @@ object CarbonDataRDDFactory {
partitionStatus: SegmentStatus = SegmentStatus.SUCCESS,
result: Option[DictionaryServer],
overwriteTable: Boolean,
+ hadoopConf: Configuration,
dataFrame: Option[DataFrame] = None,
updateModel: Option[UpdateTableModel] = None): Unit = {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
@@ -320,14 +321,14 @@ object CarbonDataRDDFactory {
}
} else {
status = if (carbonTable.getPartitionInfo(carbonTable.getTableName) != null) {
- loadDataForPartitionTable(sqlContext, dataFrame, carbonLoadModel)
+ loadDataForPartitionTable(sqlContext, dataFrame, carbonLoadModel, hadoopConf)
} else if (isSortTable && sortScope.equals(SortScopeOptions.SortScope.GLOBAL_SORT)) {
DataLoadProcessBuilderOnSpark.loadDataUsingGlobalSort(sqlContext.sparkContext,
- dataFrame, carbonLoadModel)
+ dataFrame, carbonLoadModel, hadoopConf)
} else if (dataFrame.isDefined) {
loadDataFrame(sqlContext, dataFrame, carbonLoadModel)
} else {
- loadDataFile(sqlContext, carbonLoadModel)
+ loadDataFile(sqlContext, carbonLoadModel, hadoopConf)
}
CommonUtil.mergeIndexFiles(sqlContext.sparkContext,
Seq(carbonLoadModel.getSegmentId), storePath, carbonTable, false)
@@ -783,7 +784,8 @@ object CarbonDataRDDFactory {
private def repartitionInputData(
sqlContext: SQLContext,
dataFrame: Option[DataFrame],
- carbonLoadModel: CarbonLoadModel): RDD[Row] = {
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration): RDD[Row] = {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getTableName)
val partitionColumn = partitionInfo.getColumnSchemaList.get(0).getColumnName
@@ -840,16 +842,15 @@ object CarbonDataRDDFactory {
}
} else {
// input data from csv files
- val hadoopConfiguration = new Configuration()
- CommonUtil.configureCSVInputFormat(hadoopConfiguration, carbonLoadModel)
- hadoopConfiguration.set(FileInputFormat.INPUT_DIR, carbonLoadModel.getFactFilePath)
+ CommonUtil.configureCSVInputFormat(hadoopConf, carbonLoadModel)
+ hadoopConf.set(FileInputFormat.INPUT_DIR, carbonLoadModel.getFactFilePath)
val columnCount = columns.length
new NewHadoopRDD[NullWritable, StringArrayWritable](
sqlContext.sparkContext,
classOf[CSVInputFormat],
classOf[NullWritable],
classOf[StringArrayWritable],
- hadoopConfiguration
+ hadoopConf
).map { currentRow =>
if (null == currentRow || null == currentRow._2) {
val row = new StringArrayRow(new Array[String](columnCount))
@@ -892,10 +893,11 @@ object CarbonDataRDDFactory {
private def loadDataForPartitionTable(
sqlContext: SQLContext,
dataFrame: Option[DataFrame],
- carbonLoadModel: CarbonLoadModel
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration
): Array[(String, (LoadMetadataDetails, ExecutionErrors))] = {
try {
- val rdd = repartitionInputData(sqlContext, dataFrame, carbonLoadModel)
+ val rdd = repartitionInputData(sqlContext, dataFrame, carbonLoadModel, hadoopConf)
new PartitionTableDataLoaderRDD(
sqlContext.sparkContext,
new DataLoadResultImpl(),
@@ -946,7 +948,8 @@ object CarbonDataRDDFactory {
*/
private def loadDataFile(
sqlContext: SQLContext,
- carbonLoadModel: CarbonLoadModel
+ carbonLoadModel: CarbonLoadModel,
+ hadoopConf: Configuration
): Array[(String, (LoadMetadataDetails, ExecutionErrors))] = {
/*
* when data load handle by node partition
@@ -956,20 +959,19 @@ object CarbonDataRDDFactory {
* for locally writing carbondata files(one file one block) in nodes
* 4)use NewCarbonDataLoadRDD to load data and write to carbondata files
*/
- val hadoopConfiguration = new Configuration(sqlContext.sparkContext.hadoopConfiguration)
// FileUtils will skip file which is no csv, and return all file path which split by ','
val filePaths = carbonLoadModel.getFactFilePath
- hadoopConfiguration.set(FileInputFormat.INPUT_DIR, filePaths)
- hadoopConfiguration.set(FileInputFormat.INPUT_DIR_RECURSIVE, "true")
- hadoopConfiguration.set("io.compression.codecs",
+ hadoopConf.set(FileInputFormat.INPUT_DIR, filePaths)
+ hadoopConf.set(FileInputFormat.INPUT_DIR_RECURSIVE, "true")
+ hadoopConf.set("io.compression.codecs",
"""org.apache.hadoop.io.compress.GzipCodec,
org.apache.hadoop.io.compress.DefaultCodec,
org.apache.hadoop.io.compress.BZip2Codec""".stripMargin)
- CommonUtil.configSplitMaxSize(sqlContext.sparkContext, filePaths, hadoopConfiguration)
+ CommonUtil.configSplitMaxSize(sqlContext.sparkContext, filePaths, hadoopConf)
val inputFormat = new org.apache.hadoop.mapreduce.lib.input.TextInputFormat
- val jobContext = new Job(hadoopConfiguration)
+ val jobContext = new Job(hadoopConf)
val rawSplits = inputFormat.getSplits(jobContext).toArray
val blockList = rawSplits.map { inputSplit =>
val fileSplit = inputSplit.asInstanceOf[FileSplit]
@@ -1020,7 +1022,8 @@ object CarbonDataRDDFactory {
sqlContext.sparkContext,
new DataLoadResultImpl(),
carbonLoadModel,
- blocksGroupBy
+ blocksGroupBy,
+ hadoopConf
).collect()
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
index 18343e5..b764132 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
@@ -246,6 +246,7 @@ class CarbonSource extends CreatableRelationProvider with RelationProvider
// create sink
StreamSinkFactory.createStreamTableSink(
sqlContext.sparkSession,
+ sqlContext.sparkSession.sessionState.newHadoopConf(),
carbonTable,
parameters)
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/LoadTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/LoadTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/LoadTableCommand.scala
index dfcd3cd..d805227 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/LoadTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/LoadTableCommand.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.command.management
import scala.collection.JavaConverters._
import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.conf.Configuration
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.analysis.{NoSuchTableException, UnresolvedAttribute}
@@ -97,6 +98,7 @@ case class LoadTableCommand(
// update the property with new value
carbonProperty.addProperty(CarbonCommonConstants.NUM_CORES_LOADING, numCoresLoading)
+ val hadoopConf = sparkSession.sessionState.newHadoopConf()
try {
val table = if (tableInfoOp.isDefined) {
CarbonTable.buildFromTableInfo(tableInfoOp.get)
@@ -126,7 +128,7 @@ case class LoadTableCommand(
""
} else {
FileUtils.getPaths(
- CarbonUtil.checkAndAppendHDFSUrl(factPathFromUser))
+ CarbonUtil.checkAndAppendHDFSUrl(factPathFromUser), hadoopConf)
}
carbonLoadModel.setFactFilePath(factPath)
DataLoadingUtil.buildCarbonLoadModel(
@@ -134,7 +136,8 @@ case class LoadTableCommand(
carbonProperty,
options,
optionsFinal,
- carbonLoadModel
+ carbonLoadModel,
+ hadoopConf
)
try{
@@ -174,13 +177,15 @@ case class LoadTableCommand(
carbonProperty,
carbonLoadModel,
columnar,
- partitionStatus)
+ partitionStatus,
+ hadoopConf)
} else {
loadData(
sparkSession,
carbonLoadModel,
columnar,
- partitionStatus)
+ partitionStatus,
+ hadoopConf)
}
} catch {
case CausedBy(ex: NoRetryException) =>
@@ -227,7 +232,8 @@ case class LoadTableCommand(
carbonProperty: CarbonProperties,
carbonLoadModel: CarbonLoadModel,
columnar: Boolean,
- partitionStatus: SegmentStatus): Unit = {
+ partitionStatus: SegmentStatus,
+ hadoopConf: Configuration): Unit = {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val carbonTableIdentifier = carbonTable.getAbsoluteTableIdentifier
.getCarbonTableIdentifier
@@ -295,6 +301,7 @@ case class LoadTableCommand(
partitionStatus,
server,
isOverwriteTable,
+ hadoopConf,
dataFrame,
updateModel)
}
@@ -303,7 +310,8 @@ case class LoadTableCommand(
sparkSession: SparkSession,
carbonLoadModel: CarbonLoadModel,
columnar: Boolean,
- partitionStatus: SegmentStatus): Unit = {
+ partitionStatus: SegmentStatus,
+ hadoopConf: Configuration): Unit = {
val (dictionaryDataFrame, loadDataFrame) = if (updateModel.isDefined) {
val fields = dataFrame.get.schema.fields
import org.apache.spark.sql.functions.udf
@@ -334,6 +342,7 @@ case class LoadTableCommand(
sparkSession.sqlContext,
carbonLoadModel,
carbonLoadModel.getTablePath,
+ hadoopConf,
dictionaryDataFrame)
CarbonDataRDDFactory.loadCarbonData(sparkSession.sqlContext,
carbonLoadModel,
@@ -342,6 +351,7 @@ case class LoadTableCommand(
partitionStatus,
None,
isOverwriteTable,
+ hadoopConf,
loadDataFrame,
updateModel)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
index 3f5d8c6..4cb9c6d 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
@@ -61,7 +61,8 @@ class AllDictionaryTestCase extends Spark2QueryTest with BeforeAndAfterAll {
carbonLoadModel.setDefaultTimestampFormat(CarbonProperties.getInstance().getProperty(
CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT))
- carbonLoadModel.setCsvHeaderColumns(CommonUtil.getCsvHeaderColumns(carbonLoadModel))
+ carbonLoadModel.setCsvHeaderColumns(
+ CommonUtil.getCsvHeaderColumns(carbonLoadModel, FileFactory.getConfiguration))
// Create table and metadata folders if not exist
val carbonTablePath = CarbonStorePath
.getCarbonTablePath(table.getTablePath, table.getCarbonTableIdentifier)
@@ -142,7 +143,10 @@ class AllDictionaryTestCase extends Spark2QueryTest with BeforeAndAfterAll {
val header = "id,name,city,age"
val carbonLoadModel = buildCarbonLoadModel(sampleRelation, null, header, sampleAllDictionaryFile)
GlobalDictionaryUtil.generateGlobalDictionary(
- sqlContext, carbonLoadModel, sampleRelation.carbonTable.getTablePath)
+ sqlContext,
+ carbonLoadModel,
+ sampleRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
DictionaryTestCaseUtil.
checkDictionary(sampleRelation, "city", "shenzhen")
@@ -151,10 +155,11 @@ class AllDictionaryTestCase extends Spark2QueryTest with BeforeAndAfterAll {
test("Support generate global dictionary from all dictionary files for complex type") {
val header = "deviceInformationId,channelsId,ROMSize,purchasedate,mobile,MAC,locationinfo,proddate,gamePointId,contractNumber"
val carbonLoadModel = buildCarbonLoadModel(complexRelation, null, header, complexAllDictionaryFile)
- GlobalDictionaryUtil
- .generateGlobalDictionary(sqlContext,
+ GlobalDictionaryUtil.generateGlobalDictionary(
+ sqlContext,
carbonLoadModel,
- complexRelation.carbonTable.getTablePath)
+ complexRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
DictionaryTestCaseUtil.
checkDictionary(complexRelation, "channelsId", "1650")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
index 78ae384..da5469a 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
@@ -173,7 +173,8 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
carbonLoadModel.setDefaultDateFormat(CarbonProperties.getInstance().getProperty(
CarbonCommonConstants.CARBON_DATE_FORMAT,
CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT))
- carbonLoadModel.setCsvHeaderColumns(CommonUtil.getCsvHeaderColumns(carbonLoadModel))
+ carbonLoadModel.setCsvHeaderColumns(
+ CommonUtil.getCsvHeaderColumns(carbonLoadModel, FileFactory.getConfiguration))
carbonLoadModel.setMaxColumns("100")
// Create table and metadata folders if not exist
val carbonTablePath = CarbonStorePath
@@ -197,8 +198,11 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
// load the first time
var carbonLoadModel = buildCarbonLoadModel(extComplexRelation, complexFilePath1,
header, extColDictFilePath1)
- GlobalDictionaryUtil.generateGlobalDictionary(sqlContext, carbonLoadModel,
- extComplexRelation.carbonTable.getTablePath)
+ GlobalDictionaryUtil.generateGlobalDictionary(
+ sqlContext,
+ carbonLoadModel,
+ extComplexRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
// check whether the dictionary is generated
DictionaryTestCaseUtil.checkDictionary(
extComplexRelation, "deviceInformationId", "10086")
@@ -206,8 +210,11 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
// load the second time
carbonLoadModel = buildCarbonLoadModel(extComplexRelation, complexFilePath1,
header, extColDictFilePath2)
- GlobalDictionaryUtil.generateGlobalDictionary(sqlContext, carbonLoadModel,
- extComplexRelation.carbonTable.getTablePath)
+ GlobalDictionaryUtil.generateGlobalDictionary(
+ sqlContext,
+ carbonLoadModel,
+ extComplexRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
// check the old dictionary and whether the new distinct value is generated
DictionaryTestCaseUtil.checkDictionary(
extComplexRelation, "deviceInformationId", "10086")
@@ -219,8 +226,11 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
// when csv delimiter is comma
var carbonLoadModel = buildCarbonLoadModel(extComplexRelation, complexFilePath1,
header, extColDictFilePath3)
- GlobalDictionaryUtil.generateGlobalDictionary(sqlContext, carbonLoadModel,
- extComplexRelation.carbonTable.getTablePath)
+ GlobalDictionaryUtil.generateGlobalDictionary(
+ sqlContext,
+ carbonLoadModel,
+ extComplexRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
// check whether the dictionary is generated
DictionaryTestCaseUtil.checkDictionary(
extComplexRelation, "channelsId", "1421|")
@@ -228,8 +238,11 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
// when csv delimiter is not comma
carbonLoadModel = buildCarbonLoadModel(verticalDelimiteRelation, complexFilePath2,
header2, extColDictFilePath3, "|")
- GlobalDictionaryUtil.generateGlobalDictionary(sqlContext, carbonLoadModel,
- verticalDelimiteRelation.carbonTable.getTablePath)
+ GlobalDictionaryUtil.generateGlobalDictionary(
+ sqlContext,
+ carbonLoadModel,
+ verticalDelimiteRelation.carbonTable.getTablePath,
+ FileFactory.getConfiguration)
// check whether the dictionary is generated
DictionaryTestCaseUtil.checkDictionary(
verticalDelimiteRelation, "channelsId", "1431,")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/cb6c2717/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala b/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
index 2c4d35f..ab7e059 100644
--- a/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
+++ b/streaming/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
@@ -19,6 +19,7 @@ package org.apache.carbondata.streaming
import scala.collection.JavaConverters._
+import org.apache.hadoop.conf.Configuration
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.execution.streaming.{CarbonAppendableStreamSink, Sink}
@@ -42,6 +43,7 @@ object StreamSinkFactory {
def createStreamTableSink(
sparkSession: SparkSession,
+ hadoopConf: Configuration,
carbonTable: CarbonTable,
parameters: Map[String, String]): Sink = {
validateParameters(parameters)
@@ -51,6 +53,7 @@ object StreamSinkFactory {
// build load model
val carbonLoadModel = buildCarbonLoadModelForStream(
sparkSession,
+ hadoopConf,
carbonTable,
parameters,
segmentId)
@@ -139,6 +142,7 @@ object StreamSinkFactory {
private def buildCarbonLoadModelForStream(
sparkSession: SparkSession,
+ hadoopConf: Configuration,
carbonTable: CarbonTable,
parameters: Map[String, String],
segmentId: String): CarbonLoadModel = {
@@ -156,8 +160,8 @@ object StreamSinkFactory {
carbonProperty,
parameters,
optionsFinal,
- carbonLoadModel
- )
+ carbonLoadModel,
+ hadoopConf)
carbonLoadModel.setSegmentId(segmentId)
// stream should use one pass
val dictionaryServerPort = parameters.getOrElse(