wuyunfeng commented on a change in pull request #4383:
URL: https://github.com/apache/incubator-doris/pull/4383#discussion_r477160493
##########
File path:
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/SparkEtlJobHandler.java
##########
@@ -140,93 +137,98 @@ public void submitEtlJob(long loadJobId, String
loadLabel, EtlJobConfig etlJobCo
.setAppName(String.format(ETL_JOB_NAME, loadLabel))
.setSparkHome(sparkHome)
.addAppArgs(jobConfigHdfsPath)
- .redirectError()
- .redirectOutput(new File(Config.sys_log_dir +
"/spark-submitter.log"));
+ .redirectError();
// spark configs
for (Map.Entry<String, String> entry :
resource.getSparkConfigs().entrySet()) {
launcher.setConf(entry.getKey(), entry.getValue());
}
// start app
- SparkAppHandle handle = null;
+ SparkLoadAppHandle handle = null;
State state = null;
String appId = null;
- int retry = 0;
String errMsg = "start spark app failed. error: ";
try {
- handle = launcher.startApplication(new SparkAppListener());
+ Process process = launcher.launch();
+ handle = new SparkLoadAppHandle(process);
+ handle.addListener(new SparkAppListener());
+ if (!FeConstants.runningUnitTest) {
+ SparkLauncherMonitors.LogMonitor logMonitor =
SparkLauncherMonitors.createLogMonitor(handle);
+ logMonitor.setSubmitTimeoutMs(GET_APPID_TIMEOUT_MS);
+ logMonitor.start();
+ try {
+ logMonitor.join();
+ } catch (InterruptedException e) {
+ logMonitor.interrupt();
+ throw new LoadException(errMsg + e.getMessage());
+ }
+ }
+ appId = handle.getAppId();
+ state = handle.getState();
} catch (IOException e) {
LOG.warn(errMsg, e);
throw new LoadException(errMsg + e.getMessage());
}
- while (retry++ < GET_APPID_MAX_RETRY_TIMES) {
- appId = handle.getAppId();
- if (appId != null) {
- break;
- }
-
- // check state and retry
- state = handle.getState();
- if (fromSparkState(state) == TEtlState.CANCELLED) {
- throw new LoadException(errMsg + "spark app state: " +
state.toString());
- }
- if (retry >= GET_APPID_MAX_RETRY_TIMES) {
- throw new LoadException(errMsg + "wait too much time for
getting appid. spark app state: "
- + state.toString());
- }
+ if (fromSparkState(state) == TEtlState.CANCELLED) {
+ throw new LoadException(errMsg + "spark app state: " +
state.toString() + ", loadJobId:" + loadJobId);
+ }
- // log
- if (retry % 10 == 0) {
- LOG.info("spark appid that handle get is null. load job id:
{}, state: {}, retry times: {}",
- loadJobId, state.toString(), retry);
- }
- try {
- Thread.sleep(GET_APPID_SLEEP_MS);
- } catch (InterruptedException e) {
- LOG.warn(e.getMessage());
- }
+ if (appId == null) {
+ throw new LoadException(errMsg + "Failed to get appId from handle.
spark app state: "
+ + state.toString() + ", loadJobId:" + loadJobId);
}
// success
attachment.setAppId(appId);
attachment.setHandle(handle);
}
- public EtlStatus getEtlJobStatus(SparkAppHandle handle, String appId, long
loadJobId, String etlOutputPath,
- SparkResource resource, BrokerDesc
brokerDesc) {
+ public EtlStatus getEtlJobStatus(SparkLoadAppHandle handle, String appId,
long loadJobId, String etlOutputPath,
+ SparkResource resource, BrokerDesc
brokerDesc) throws LoadException {
EtlStatus status = new EtlStatus();
if (resource.isYarnMaster()) {
- // state from yarn
Preconditions.checkState(appId != null && !appId.isEmpty());
Review comment:
Maybe put statement `Preconditions.checkState ` outer the if statement?
##########
File path: fe/fe-core/src/main/java/org/apache/doris/catalog/SparkResource.java
##########
@@ -243,6 +264,16 @@ protected void setProperties(Map<String, String>
properties) throws DdlException
return sparkConfigs;
}
+ private Map<String, String> getSparkHadoopConfigs(Map<String, String>
properties) {
+ Map<String, String> sparkConfigs = Maps.newHashMap();
Review comment:
```suggestion
Map<String, String> sparkConfig = Maps.newHashMap();
```
##########
File path:
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/SparkLauncherMonitors.java
##########
@@ -0,0 +1,231 @@
+// 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.doris.load.loadv2;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Splitter;
+import com.google.common.base.Strings;
+
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class SparkLauncherMonitors {
Review comment:
I do not like the `xxxxs`
##########
File path: fe/fe-core/src/main/java/org/apache/doris/load/loadv2/ConfigFile.java
##########
@@ -0,0 +1,25 @@
+// 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.doris.load.loadv2;
+
+import org.apache.doris.common.LoadException;
+
+public interface ConfigFile {
Review comment:
Can you add some comment ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
