Skip to content

Commit 2e87fb4

Browse files
author
gaodayue
committed
KYLIN-2165 fix IT
1 parent 93cc5ab commit 2e87fb4

File tree

4 files changed

+58
-75
lines changed

4 files changed

+58
-75
lines changed

‎core-job/src/main/java/org/apache/kylin/job/JoinedFlatTable.java

+17-15
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,12 @@
2020

2121
import java.io.File;
2222
import java.util.HashSet;
23+
import java.util.Map;
2324
import java.util.Set;
2425

2526
import javax.xml.parsers.DocumentBuilder;
2627
import javax.xml.parsers.DocumentBuilderFactory;
2728

28-
import org.apache.kylin.job.engine.JobEngineConfig;
2929
import org.apache.kylin.metadata.model.DataModelDesc;
3030
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
3131
import org.apache.kylin.metadata.model.JoinDesc;
@@ -46,30 +46,35 @@ public static String getTableDir(IJoinedFlatTableDesc flatDesc, String storageDf
4646
return storageDfsDir + "/" + flatDesc.getTableName();
4747
}
4848

49-
public static String generateHiveSetStatements(JobEngineConfig engineConfig) {
49+
public static String generateHiveInitStatements(
50+
String flatTableDatabase, String kylinHiveFile, Map<String, String> cubeOverrides) {
51+
5052
StringBuilder buffer = new StringBuilder();
5153

54+
buffer.append("USE ").append(flatTableDatabase).append(";\n");
5255
try {
53-
File hadoopPropertiesFile = new File(engineConfig.getHiveConfFilePath());
54-
55-
if (hadoopPropertiesFile.exists()) {
56+
File file = new File(kylinHiveFile);
57+
if (file.exists()) {
5658
DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
57-
DocumentBuilder builder;
58-
Document doc;
59-
builder = factory.newDocumentBuilder();
60-
doc = builder.parse(hadoopPropertiesFile);
59+
DocumentBuilder builder = factory.newDocumentBuilder();
60+
Document doc = builder.parse(file);
6161
NodeList nl = doc.getElementsByTagName("property");
6262
for (int i = 0; i < nl.getLength(); i++) {
6363
String name = doc.getElementsByTagName("name").item(i).getFirstChild().getNodeValue();
6464
String value = doc.getElementsByTagName("value").item(i).getFirstChild().getNodeValue();
6565
if (!name.equals("tmpjars")) {
66-
buffer.append("SET " + name + "=" + value + ";\n");
66+
buffer.append("SET ").append(name).append("=").append(value).append(";\n");
6767
}
6868
}
6969
}
7070
} catch (Exception e) {
7171
throw new RuntimeException("Failed to parse hive conf file ", e);
7272
}
73+
74+
for (Map.Entry<String, String> entry : cubeOverrides.entrySet()) {
75+
buffer.append("SET ").append(entry.getKey()).append("=").append(entry.getValue()).append(";\n");
76+
}
77+
7378
return buffer.toString();
7479
}
7580

@@ -98,11 +103,8 @@ public static String generateDropTableStatement(IJoinedFlatTableDesc flatDesc) {
98103
return ddl.toString();
99104
}
100105

101-
public static String generateInsertDataStatement(IJoinedFlatTableDesc flatDesc, JobEngineConfig engineConfig) {
102-
StringBuilder sql = new StringBuilder();
103-
sql.append(generateHiveSetStatements(engineConfig));
104-
sql.append("INSERT OVERWRITE TABLE " + flatDesc.getTableName() + " " + generateSelectDataStatement(flatDesc) + ";").append("\n");
105-
return sql.toString();
106+
public static String generateInsertDataStatement(IJoinedFlatTableDesc flatDesc) {
107+
return "INSERT OVERWRITE TABLE " + flatDesc.getTableName() + " " + generateSelectDataStatement(flatDesc) + ";\n";
106108
}
107109

108110
public static String generateSelectDataStatement(IJoinedFlatTableDesc flatDesc) {

‎core-job/src/main/java/org/apache/kylin/job/engine/JobEngineConfig.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ public String getHadoopJobConfFilePath(String jobType) throws IOException {
9595
return path;
9696
}
9797

98-
public String getHiveConfFilePath() throws IOException {
98+
public String getHiveConfFilePath() {
9999
String hiveConfFile = (HIVE_CONF_FILENAME + ".xml");
100100

101101
File jobConfig = getJobConfig(hiveConfFile);

‎core-job/src/test/java/org/apache/kylin/job/JoinedFlatTableTest.java

+1-3
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,11 @@
2222

2323
import java.io.IOException;
2424

25-
import org.apache.kylin.common.KylinConfig;
2625
import org.apache.kylin.common.util.LocalFileMetadataTestCase;
2726
import org.apache.kylin.cube.CubeInstance;
2827
import org.apache.kylin.cube.CubeManager;
2928
import org.apache.kylin.cube.CubeSegment;
3029
import org.apache.kylin.engine.EngineFactory;
31-
import org.apache.kylin.job.engine.JobEngineConfig;
3230
import org.apache.kylin.metadata.model.IJoinedFlatTableDesc;
3331
import org.junit.After;
3432
import org.junit.Before;
@@ -77,7 +75,7 @@ public void testGenDropTableDDL() {
7775

7876
@Test
7977
public void testGenerateInsertSql() throws IOException {
80-
String sqls = JoinedFlatTable.generateInsertDataStatement(flatTableDesc, new JobEngineConfig(KylinConfig.getInstanceFromEnv()));
78+
String sqls = JoinedFlatTable.generateInsertDataStatement(flatTableDesc);
8179
System.out.println(sqls);
8280

8381
int length = sqls.length();

‎source-hive/src/main/java/org/apache/kylin/source/hive/HiveMRInput.java

+39-56
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.apache.kylin.source.hive;
2020

2121
import java.io.IOException;
22-
import java.util.Map;
2322
import java.util.Set;
2423

2524
import org.apache.commons.lang.StringUtils;
@@ -127,51 +126,58 @@ public String[] parseMapperInput(Object mapperInput) {
127126

128127
public static class BatchCubingInputSide implements IMRBatchCubingInputSide {
129128

130-
JobEngineConfig conf;
131129
final IJoinedFlatTableDesc flatDesc;
130+
final String flatTableDatabase;
131+
final String hdfsWorkingDir;
132+
132133
String hiveViewIntermediateTables = "";
133134

134135
public BatchCubingInputSide(IJoinedFlatTableDesc flatDesc) {
136+
KylinConfig config = KylinConfig.getInstanceFromEnv();
135137
this.flatDesc = flatDesc;
138+
this.flatTableDatabase = config.getHiveDatabaseForIntermediateTable();
139+
this.hdfsWorkingDir = config.getHdfsWorkingDirectory();
136140
}
137141

138142
@Override
139143
public void addStepPhase1_CreateFlatTable(DefaultChainedExecutable jobFlow) {
140144
final String cubeName = CubingExecutableUtil.getCubeName(jobFlow.getParams());
141-
final KylinConfig kylinConfig = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()).getCube(cubeName).getConfig();
142-
this.conf = new JobEngineConfig(kylinConfig);
145+
final KylinConfig cubeConfig = CubeManager.getInstance(KylinConfig.getInstanceFromEnv()).getCube(cubeName).getConfig();
146+
JobEngineConfig conf = new JobEngineConfig(cubeConfig);
147+
148+
final String hiveInitStatements = JoinedFlatTable.generateHiveInitStatements(
149+
flatTableDatabase, conf.getHiveConfFilePath(), cubeConfig.getHiveConfigOverride()
150+
) ;
151+
final String jobWorkingDir = getJobWorkingDir(jobFlow);
143152

144153
// create flat table first, then count and redistribute
145-
jobFlow.addTask(createFlatHiveTableStep(conf, flatDesc, jobFlow.getId(), cubeName));
146-
if (kylinConfig.isHiveRedistributeEnabled() == true) {
147-
jobFlow.addTask(createRedistributeFlatHiveTableStep(conf, flatDesc, jobFlow.getId(), cubeName));
154+
jobFlow.addTask(createFlatHiveTableStep(hiveInitStatements, jobWorkingDir, cubeName));
155+
if (cubeConfig.isHiveRedistributeEnabled() == true) {
156+
jobFlow.addTask(createRedistributeFlatHiveTableStep(hiveInitStatements, cubeName));
148157
}
149-
AbstractExecutable task = createLookupHiveViewMaterializationStep(jobFlow.getId());
158+
AbstractExecutable task = createLookupHiveViewMaterializationStep(hiveInitStatements, jobWorkingDir);
150159
if (task != null) {
151160
jobFlow.addTask(task);
152161
}
153162
}
154163

155-
public static AbstractExecutable createRedistributeFlatHiveTableStep(JobEngineConfig conf, IJoinedFlatTableDesc flatTableDesc, String jobId, String cubeName) {
156-
StringBuilder hiveInitBuf = new StringBuilder();
157-
hiveInitBuf.append("USE ").append(conf.getConfig().getHiveDatabaseForIntermediateTable()).append(";\n");
158-
hiveInitBuf.append(JoinedFlatTable.generateHiveSetStatements(conf));
159-
final KylinConfig kylinConfig = ((CubeSegment) flatTableDesc.getSegment()).getConfig();
160-
appendHiveOverrideProperties(kylinConfig, hiveInitBuf);
164+
private String getJobWorkingDir(DefaultChainedExecutable jobFlow) {
165+
return JobBuilderSupport.getJobWorkingDir(hdfsWorkingDir, jobFlow.getId());
166+
}
161167

168+
private AbstractExecutable createRedistributeFlatHiveTableStep(String hiveInitStatements, String cubeName) {
162169
RedistributeFlatHiveTableStep step = new RedistributeFlatHiveTableStep();
163-
step.setInitStatement(hiveInitBuf.toString());
164-
step.setIntermediateTable(flatTableDesc.getTableName());
165-
step.setRedistributeDataStatement(JoinedFlatTable.generateRedistributeFlatTableStatement(flatTableDesc));
170+
step.setInitStatement(hiveInitStatements);
171+
step.setIntermediateTable(flatDesc.getTableName());
172+
step.setRedistributeDataStatement(JoinedFlatTable.generateRedistributeFlatTableStatement(flatDesc));
166173
CubingExecutableUtil.setCubeName(cubeName, step.getParams());
167174
step.setName(ExecutableConstants.STEP_NAME_REDISTRIBUTE_FLAT_HIVE_TABLE);
168175
return step;
169176
}
170177

171-
public ShellExecutable createLookupHiveViewMaterializationStep(String jobId) {
178+
private ShellExecutable createLookupHiveViewMaterializationStep(String hiveInitStatements, String jobWorkingDir) {
172179
ShellExecutable step = new ShellExecutable();
173180
step.setName(ExecutableConstants.STEP_NAME_MATERIALIZE_HIVE_VIEW_IN_LOOKUP);
174-
HiveCmdBuilder hiveCmdBuilder = new HiveCmdBuilder();
175181

176182
KylinConfig kylinConfig = ((CubeSegment) flatDesc.getSegment()).getConfig();
177183
MetadataManager metadataManager = MetadataManager.getInstance(kylinConfig);
@@ -187,16 +193,15 @@ public ShellExecutable createLookupHiveViewMaterializationStep(String jobId) {
187193
if (lookupViewsTables.size() == 0) {
188194
return null;
189195
}
190-
appendHiveOverrideProperties2(kylinConfig, hiveCmdBuilder);
191-
final String useDatabaseHql = "USE " + conf.getConfig().getHiveDatabaseForIntermediateTable() + ";";
192-
hiveCmdBuilder.addStatement(useDatabaseHql);
193-
hiveCmdBuilder.addStatement(JoinedFlatTable.generateHiveSetStatements(conf));
196+
197+
HiveCmdBuilder hiveCmdBuilder = new HiveCmdBuilder();
198+
hiveCmdBuilder.addStatement(hiveInitStatements);
194199
for (TableDesc lookUpTableDesc : lookupViewsTables) {
195200
if (lookUpTableDesc.isView()) {
196201
StringBuilder createIntermediateTableHql = new StringBuilder();
197202
createIntermediateTableHql.append("DROP TABLE IF EXISTS " + lookUpTableDesc.getMaterializedName() + ";\n");
198203
createIntermediateTableHql.append("CREATE TABLE IF NOT EXISTS " + lookUpTableDesc.getMaterializedName() + "\n");
199-
createIntermediateTableHql.append("LOCATION '" + JobBuilderSupport.getJobWorkingDir(conf, jobId) + "/" + lookUpTableDesc.getMaterializedName() + "'\n");
204+
createIntermediateTableHql.append("LOCATION '" + jobWorkingDir + "/" + lookUpTableDesc.getMaterializedName() + "'\n");
200205
createIntermediateTableHql.append("AS SELECT * FROM " + lookUpTableDesc.getIdentity() + ";\n");
201206
hiveCmdBuilder.addStatement(createIntermediateTableHql.toString());
202207
hiveViewIntermediateTables = hiveViewIntermediateTables + lookUpTableDesc.getMaterializedName() + ";";
@@ -209,30 +214,27 @@ public ShellExecutable createLookupHiveViewMaterializationStep(String jobId) {
209214
return step;
210215
}
211216

212-
public static AbstractExecutable createFlatHiveTableStep(JobEngineConfig conf, IJoinedFlatTableDesc flatTableDesc, String jobId, String cubeName) {
213-
StringBuilder hiveInitBuf = new StringBuilder();
214-
hiveInitBuf.append(JoinedFlatTable.generateHiveSetStatements(conf));
215-
final KylinConfig kylinConfig = ((CubeSegment) flatTableDesc.getSegment()).getConfig();
216-
appendHiveOverrideProperties(kylinConfig, hiveInitBuf);
217-
final String useDatabaseHql = "USE " + conf.getConfig().getHiveDatabaseForIntermediateTable() + ";\n";
218-
final String dropTableHql = JoinedFlatTable.generateDropTableStatement(flatTableDesc);
219-
final String createTableHql = JoinedFlatTable.generateCreateTableStatement(flatTableDesc, JobBuilderSupport.getJobWorkingDir(conf, jobId));
220-
String insertDataHqls = JoinedFlatTable.generateInsertDataStatement(flatTableDesc, conf);
217+
private AbstractExecutable createFlatHiveTableStep(String hiveInitStatements, String jobWorkingDir, String cubeName) {
218+
final String dropTableHql = JoinedFlatTable.generateDropTableStatement(flatDesc);
219+
final String createTableHql = JoinedFlatTable.generateCreateTableStatement(flatDesc, jobWorkingDir);
220+
String insertDataHqls = JoinedFlatTable.generateInsertDataStatement(flatDesc);
221221

222222
CreateFlatHiveTableStep step = new CreateFlatHiveTableStep();
223-
step.setInitStatement(hiveInitBuf.toString());
224-
step.setCreateTableStatement(useDatabaseHql + dropTableHql + createTableHql + insertDataHqls);
223+
step.setInitStatement(hiveInitStatements);
224+
step.setCreateTableStatement(dropTableHql + createTableHql + insertDataHqls);
225225
CubingExecutableUtil.setCubeName(cubeName, step.getParams());
226226
step.setName(ExecutableConstants.STEP_NAME_CREATE_FLAT_HIVE_TABLE);
227227
return step;
228228
}
229229

230230
@Override
231231
public void addStepPhase4_Cleanup(DefaultChainedExecutable jobFlow) {
232+
final String jobWorkingDir = getJobWorkingDir(jobFlow);
233+
232234
GarbageCollectionStep step = new GarbageCollectionStep();
233235
step.setName(ExecutableConstants.STEP_NAME_HIVE_CLEANUP);
234236
step.setIntermediateTableIdentity(getIntermediateTableIdentity());
235-
step.setExternalDataPath(JoinedFlatTable.getTableDir(flatDesc, JobBuilderSupport.getJobWorkingDir(conf, jobFlow.getId())));
237+
step.setExternalDataPath(JoinedFlatTable.getTableDir(flatDesc, jobWorkingDir));
236238
step.setHiveViewIntermediateTableIdentities(hiveViewIntermediateTables);
237239
jobFlow.addTask(step);
238240
}
@@ -243,7 +245,7 @@ public IMRTableInputFormat getFlatTableInputFormat() {
243245
}
244246

245247
private String getIntermediateTableIdentity() {
246-
return conf.getConfig().getHiveDatabaseForIntermediateTable() + "." + flatDesc.getTableName();
248+
return flatTableDatabase + "." + flatDesc.getTableName();
247249
}
248250
}
249251

@@ -413,24 +415,5 @@ private String getExternalDataPath() {
413415
public void setHiveViewIntermediateTableIdentities(String tableIdentities) {
414416
setParam("oldHiveViewIntermediateTables", tableIdentities);
415417
}
416-
417-
}
418-
419-
private static void appendHiveOverrideProperties(final KylinConfig kylinConfig, StringBuilder hiveCmd) {
420-
final Map<String, String> hiveConfOverride = kylinConfig.getHiveConfigOverride();
421-
if (hiveConfOverride.isEmpty() == false) {
422-
for (String key : hiveConfOverride.keySet()) {
423-
hiveCmd.append("SET ").append(key).append("=").append(hiveConfOverride.get(key)).append(";\n");
424-
}
425-
}
426-
}
427-
428-
private static void appendHiveOverrideProperties2(final KylinConfig kylinConfig, HiveCmdBuilder hiveCmdBuilder) {
429-
final Map<String, String> hiveConfOverride = kylinConfig.getHiveConfigOverride();
430-
if (hiveConfOverride.isEmpty() == false) {
431-
for (String key : hiveConfOverride.keySet()) {
432-
hiveCmdBuilder.addStatement("SET " + key + "=" + hiveConfOverride.get(key) + ";\n");
433-
}
434-
}
435418
}
436419
}

0 commit comments

Comments
 (0)