Wayne1c closed pull request #391: KYLIN-3699 SparkCubing failed when build with
empty data
URL: https://github.com/apache/kylin/pull/391
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core-job/src/main/java/org/apache/kylin/job/JoinedFlatTable.java
b/core-job/src/main/java/org/apache/kylin/job/JoinedFlatTable.java
index ddd883d645..3383f54b43 100644
--- a/core-job/src/main/java/org/apache/kylin/job/JoinedFlatTable.java
+++ b/core-job/src/main/java/org/apache/kylin/job/JoinedFlatTable.java
@@ -92,6 +92,11 @@ public static String
generateCreateTableStatement(IJoinedFlatTableDesc flatDesc,
ddl.append(colName(col, flatDesc.useAlias()) + " " +
getHiveDataType(col.getDatatype()) + "\n");
}
ddl.append(")" + "\n");
+
+ if (!("TEXTFILE".equals(storageFormat.toUpperCase(Locale.ROOT)) ||
"SEQUENCEFILE".equals(storageFormat.toUpperCase(Locale.ROOT)))) {
+ throw new RuntimeException("Invalid storage format of flat table:
" + storageFormat.toUpperCase(Locale.ROOT));
+ }
+
if (TEXTFILE.equals(storageFormat)) {
ddl.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY '" +
fieldDelimiter + "'\n");
}
diff --git
a/source-hive/src/main/java/org/apache/kylin/source/hive/CreateFlatHiveTableStep.java
b/source-hive/src/main/java/org/apache/kylin/source/hive/CreateFlatHiveTableStep.java
index 21ef940003..b6b1f42fc1 100644
---
a/source-hive/src/main/java/org/apache/kylin/source/hive/CreateFlatHiveTableStep.java
+++
b/source-hive/src/main/java/org/apache/kylin/source/hive/CreateFlatHiveTableStep.java
@@ -18,15 +18,21 @@
package org.apache.kylin.source.hive;
import java.io.IOException;
+import java.util.Locale;
import java.util.Map;
import java.util.regex.Pattern;
import java.util.regex.Matcher;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.Text;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.common.util.HiveCmdBuilder;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeInstance;
@@ -97,6 +103,7 @@ protected ExecuteResult doWork(ExecutableContext context)
throws ExecuteExceptio
KylinConfig config = getCubeSpecificConfig();
try {
createFlatHiveTable(config);
+ checkAndAppendFlatTableFile();
return new ExecuteResult(ExecuteResult.State.SUCCEED,
stepLogger.getBufferedLog());
} catch (Exception e) {
@@ -105,6 +112,57 @@ protected ExecuteResult doWork(ExecutableContext context)
throws ExecuteExceptio
}
}
+ public void checkAndAppendFlatTableFile() throws IOException {
+ FileSystem fs = HadoopUtil.getWorkingFileSystem();
+ if (getFlatTableName() == null || getFlatTableStorageFormat() == null)
{
+ return;
+ }
+
+ Path path = new Path(getWorkingDir(), getFlatTableName());
+
+ logger.debug("Checking flat table files in " + path);
+ FileStatus[] fileStatus = fs.listStatus(path);
+ if (fileStatus.length > 0)
+ return;
+
+ logger.debug("Append empty flat table file in " + path);
+ String flatTableStorageFormat =
getFlatTableStorageFormat().toUpperCase(Locale.ROOT);
+ if ("TEXTFILE".equals(flatTableStorageFormat)) {
+ fs.create(new Path(path, "part-m-00000")).close();
+ } else if ("SEQUENCEFILE".equals(flatTableStorageFormat)) {
+ SequenceFile.createWriter(HadoopUtil.getCurrentConfiguration(),
+ SequenceFile.Writer.file(new Path(path, "000000_0")),
+ SequenceFile.Writer.keyClass(BytesWritable.class),
SequenceFile.Writer.valueClass(Text.class))
+ .close();
+ } else {
+ throw new RuntimeException("Invalid storage format of flat table:
" + flatTableStorageFormat);
+ }
+ }
+
+ public void setFlatTableName(String storageFormat) {
+ setParam("FlatTableName", storageFormat);
+ }
+
+ public String getFlatTableName() {
+ return getParam("FlatTableName");
+ }
+
+ public void setFlatTableStorageFormat(String storageFormat) {
+ setParam("FlatTableStorageFormat", storageFormat);
+ }
+
+ public String getFlatTableStorageFormat() {
+ return getParam("FlatTableStorageFormat");
+ }
+
+ public void setWorkingDir(String workingDir) {
+ setParam("WorkingDir", workingDir);
+ }
+
+ public String getWorkingDir() {
+ return getParam("WorkingDir");
+ }
+
public void setInitStatement(String sql) {
setParam("HiveInit", sql);
}
diff --git
a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
index 2f25e50016..be3866979c 100644
--- a/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
+++ b/source-hive/src/main/java/org/apache/kylin/source/hive/HiveInputBase.java
@@ -149,6 +149,9 @@ protected static AbstractExecutable
createFlatHiveTableStep(String hiveInitState
step.setCreateTableStatement(dropTableHql + createTableHql +
insertDataHqls);
CubingExecutableUtil.setCubeName(cubeName, step.getParams());
step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
+ step.setWorkingDir(jobWorkingDir);
+
step.setFlatTableStorageFormat(flatDesc.getDataModel().getConfig().getFlatTableStorageFormat());
+ step.setFlatTableName(flatDesc.getTableName());
return step;
}
diff --git
a/source-hive/src/test/java/org/apache/kylin/source/hive/CreateFlatHiveTableStepTest.java
b/source-hive/src/test/java/org/apache/kylin/source/hive/CreateFlatHiveTableStepTest.java
new file mode 100644
index 0000000000..e9271acb64
--- /dev/null
+++
b/source-hive/src/test/java/org/apache/kylin/source/hive/CreateFlatHiveTableStepTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.kylin.source.hive;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CreateFlatHiveTableStepTest extends LocalFileMetadataTestCase {
+
+ @Before
+ public void setUp() throws Exception {
+ this.createTestMetadata();
+ }
+
+ @After
+ public void after() throws Exception {
+ this.cleanupTestMetadata();
+ }
+
+ @Test
+ public void testCheckAndAppendFlatTableFile() throws IOException {
+ FileSystem fileSystem = FileSystem.get(new Configuration());
+ String absolutePath = new File("./hivedir/").getAbsolutePath();
+ if (absolutePath.startsWith("/"))
+ absolutePath = "file://" + absolutePath;
+ else
+ absolutePath = "file:///" + absolutePath;
+ String flatTableName =
"kylin_intermediate_cube_937e983f_6ae9_4d88_8528_3553c8f13026";
+ Path jobWorkDirPath = new Path(absolutePath, flatTableName);
+
+ CreateFlatHiveTableStep createFlatHiveTableStep = new
CreateFlatHiveTableStep();
+ createFlatHiveTableStep.setWorkingDir(absolutePath);
+ try {
+ fileSystem.mkdirs(jobWorkDirPath);
+ createFlatHiveTableStep.setFlatTableStorageFormat("SequenceFile");
+ createFlatHiveTableStep.setFlatTableName(flatTableName);
+ createFlatHiveTableStep.checkAndAppendFlatTableFile();
+ Assert.assertTrue(fileSystem.exists(new Path(jobWorkDirPath,
"000000_0")));
+ } finally {
+ if (jobWorkDirPath != null)
+ fileSystem.deleteOnExit(new Path(absolutePath));
+ }
+
+ }
+
+ @Test
+ public void testSetWorkingDir() {
+ CreateFlatHiveTableStep createFlatHiveTableStep = new
CreateFlatHiveTableStep();
+ String jobWorkingDir = "/tmp/kylin";
+ createFlatHiveTableStep.setWorkingDir(jobWorkingDir);
+ Assert.assertEquals(createFlatHiveTableStep.getWorkingDir(),
jobWorkingDir);
+ }
+
+ @Test
+ public void testSetCreateTableStatement() {
+ CreateFlatHiveTableStep createFlatHiveTableStep = new
CreateFlatHiveTableStep();
+ String createTableStatement = "HiveRedistributeData";
+ createFlatHiveTableStep.setCreateTableStatement(createTableStatement);
+ Assert.assertEquals(createFlatHiveTableStep.getCreateTableStatement(),
createTableStatement);
+ }
+
+ @Test
+ public void testSetInitStatement() {
+ CreateFlatHiveTableStep createFlatHiveTableStep = new
CreateFlatHiveTableStep();
+ String initStatement = "HiveInit";
+ createFlatHiveTableStep.setInitStatement(initStatement);
+ Assert.assertEquals(createFlatHiveTableStep.getInitStatement(),
initStatement);
+ }
+
+}
\ No newline at end of file
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
