You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2020/05/28 16:55:53 UTC
[GitHub] [incubator-doris] wyb opened a new pull request #3716: [Spark load] Fe submit spark etl job
wyb opened a new pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716
After user creates a spark load job which status is PENDING, Fe will schedule and submit the spark etl job.
1. Begin transaction
2. Create a SparkLoadPendingTask for submitting etl job
2.1 Create etl job configuration according to https://github.com/apache/incubator-doris/issues/3010#issuecomment-635174675
2.2 Upload the configuration file and job jar to HDFS with broker
2.3 Submit etl job to spark cluster
2.4 Wait for etl job submission result
3. Update job state to ETL and log job update info if etl job is submitted successfully
#3433
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wyb commented on a change in pull request #3716: [Spark load][Fe 4/5] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
wyb commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r433351923
##########
File path: fe/src/main/java/org/apache/doris/load/loadv2/SparkEtlJobHandler.java
##########
@@ -0,0 +1,170 @@
+// 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.PaloFe;
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.catalog.SparkResource;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.BrokerUtil;
+import org.apache.doris.load.loadv2.etl.EtlJobConfig;
+import org.apache.doris.thrift.TEtlState;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.spark.launcher.SparkAppHandle;
+import org.apache.spark.launcher.SparkAppHandle.Listener;
+import org.apache.spark.launcher.SparkAppHandle.State;
+import org.apache.spark.launcher.SparkLauncher;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * SparkEtlJobHandler is responsible for
+ * 1. submit spark etl job
+ * 2. get spark etl job status
+ * 3. kill spark etl job
+ * 4. get spark etl file paths
+ * 5. delete etl output path
+ */
+public class SparkEtlJobHandler {
+ private static final Logger LOG = LogManager.getLogger(SparkEtlJobHandler.class);
+
+ private static final String APP_RESOURCE_NAME = "palo-fe.jar";
+ private static final String CONFIG_FILE_NAME = "jobconfig.json";
+ private static final String APP_RESOURCE_LOCAL_PATH = PaloFe.DORIS_HOME_DIR + "/lib/" + APP_RESOURCE_NAME;
+ private static final String JOB_CONFIG_DIR = "configs";
+ private static final String MAIN_CLASS = "org.apache.doris.load.loadv2.etl.SparkEtlJob";
Review comment:
ok, I comment and will replace with it when SparkEtlJob class is merged.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman commented on a change in pull request #3716: [Spark load] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
morningman commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r432800096
##########
File path: fe/src/main/java/org/apache/doris/catalog/OlapTable.java
##########
@@ -530,6 +530,30 @@ public KeysType getKeysType() {
return keysType;
}
+ public KeysType getKeysTypeByIndexId(long indexId) {
Review comment:
There is `keysType` is `MaterializedIndexMeta`. You can get it directly.
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -17,52 +17,59 @@
package org.apache.doris.common.util;
-import com.google.common.collect.Lists;
import org.apache.doris.analysis.BrokerDesc;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.catalog.FsBroker;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.apache.doris.service.FrontendOptions;
+import org.apache.doris.thrift.TBrokerCloseReaderRequest;
+import org.apache.doris.thrift.TBrokerCloseWriterRequest;
+import org.apache.doris.thrift.TBrokerDeletePathRequest;
+import org.apache.doris.thrift.TBrokerFD;
import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TBrokerListPathRequest;
import org.apache.doris.thrift.TBrokerListResponse;
+import org.apache.doris.thrift.TBrokerOpenMode;
+import org.apache.doris.thrift.TBrokerOpenReaderRequest;
+import org.apache.doris.thrift.TBrokerOpenReaderResponse;
+import org.apache.doris.thrift.TBrokerOpenWriterRequest;
+import org.apache.doris.thrift.TBrokerOpenWriterResponse;
+import org.apache.doris.thrift.TBrokerOperationStatus;
import org.apache.doris.thrift.TBrokerOperationStatusCode;
+import org.apache.doris.thrift.TBrokerPReadRequest;
+import org.apache.doris.thrift.TBrokerPWriteRequest;
+import org.apache.doris.thrift.TBrokerReadResponse;
import org.apache.doris.thrift.TBrokerVersion;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPaloBrokerService;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
import java.util.Collections;
import java.util.List;
public class BrokerUtil {
private static final Logger LOG = LogManager.getLogger(BrokerUtil.class);
+ private static int READ_BUFFER_SIZE = 1024 * 1024;
Review comment:
add unit to name.
##########
File path: fe/src/main/java/org/apache/doris/common/Pair.java
##########
@@ -25,7 +27,9 @@
public class Pair<F, S> {
public static PairComparator<Pair<?, Comparable>> PAIR_VALUE_COMPARATOR = new PairComparator<>();
+ @SerializedName(value = "first")
Review comment:
I'am not sure this is ok, cause there is no guarantee that the `F` and `S` object can also be serialized by GSON
##########
File path: fe/src/main/java/org/apache/doris/load/loadv2/SparkEtlJobHandler.java
##########
@@ -0,0 +1,170 @@
+// 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.PaloFe;
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.catalog.SparkResource;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.BrokerUtil;
+import org.apache.doris.load.loadv2.etl.EtlJobConfig;
+import org.apache.doris.thrift.TEtlState;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.spark.launcher.SparkAppHandle;
+import org.apache.spark.launcher.SparkAppHandle.Listener;
+import org.apache.spark.launcher.SparkAppHandle.State;
+import org.apache.spark.launcher.SparkLauncher;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * SparkEtlJobHandler is responsible for
+ * 1. submit spark etl job
+ * 2. get spark etl job status
+ * 3. kill spark etl job
+ * 4. get spark etl file paths
+ * 5. delete etl output path
+ */
+public class SparkEtlJobHandler {
+ private static final Logger LOG = LogManager.getLogger(SparkEtlJobHandler.class);
+
+ private static final String APP_RESOURCE_NAME = "palo-fe.jar";
+ private static final String CONFIG_FILE_NAME = "jobconfig.json";
+ private static final String APP_RESOURCE_LOCAL_PATH = PaloFe.DORIS_HOME_DIR + "/lib/" + APP_RESOURCE_NAME;
+ private static final String JOB_CONFIG_DIR = "configs";
+ private static final String MAIN_CLASS = "org.apache.doris.load.loadv2.etl.SparkEtlJob";
Review comment:
How about get it from `SparkEtlJob.class.getXXX()`?
##########
File path: fe/src/main/java/org/apache/doris/load/loadv2/SparkEtlJobHandler.java
##########
@@ -0,0 +1,170 @@
+// 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.PaloFe;
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.catalog.SparkResource;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.BrokerUtil;
+import org.apache.doris.load.loadv2.etl.EtlJobConfig;
+import org.apache.doris.thrift.TEtlState;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.spark.launcher.SparkAppHandle;
+import org.apache.spark.launcher.SparkAppHandle.Listener;
+import org.apache.spark.launcher.SparkAppHandle.State;
+import org.apache.spark.launcher.SparkLauncher;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * SparkEtlJobHandler is responsible for
+ * 1. submit spark etl job
+ * 2. get spark etl job status
+ * 3. kill spark etl job
+ * 4. get spark etl file paths
+ * 5. delete etl output path
+ */
+public class SparkEtlJobHandler {
+ private static final Logger LOG = LogManager.getLogger(SparkEtlJobHandler.class);
+
+ private static final String APP_RESOURCE_NAME = "palo-fe.jar";
+ private static final String CONFIG_FILE_NAME = "jobconfig.json";
+ private static final String APP_RESOURCE_LOCAL_PATH = PaloFe.DORIS_HOME_DIR + "/lib/" + APP_RESOURCE_NAME;
+ private static final String JOB_CONFIG_DIR = "configs";
+ private static final String MAIN_CLASS = "org.apache.doris.load.loadv2.etl.SparkEtlJob";
+ private static final String ETL_JOB_NAME = "doris__%s";
+ // 5min
+ private static final int GET_APPID_MAX_RETRY_TIMES = 300;
+ private static final int GET_APPID_SLEEP_MS = 1000;
+
+ class SparkAppListener implements Listener {
+ @Override
+ public void stateChanged(SparkAppHandle sparkAppHandle) {}
+
+ @Override
+ public void infoChanged(SparkAppHandle sparkAppHandle) {}
+ }
+
+ public void submitEtlJob(long loadJobId, String loadLabel, EtlJobConfig etlJobConfig, SparkResource resource,
+ BrokerDesc brokerDesc, SparkPendingTaskAttachment attachment) throws LoadException {
+ // delete outputPath
+ deleteEtlOutputPath(etlJobConfig.outputPath, brokerDesc);
+
+ // upload app resource and jobconfig to hdfs
+ String configsHdfsDir = etlJobConfig.outputPath + "/" + JOB_CONFIG_DIR + "/";
+ String appResourceHdfsPath = configsHdfsDir + APP_RESOURCE_NAME;
+ String jobConfigHdfsPath = configsHdfsDir + CONFIG_FILE_NAME;
+ try {
+ BrokerUtil.writeBrokerFile(APP_RESOURCE_LOCAL_PATH, appResourceHdfsPath, brokerDesc);
+ byte[] configData = etlJobConfig.configToJson().getBytes("UTF-8");
+ BrokerUtil.writeBrokerFile(configData, jobConfigHdfsPath, brokerDesc);
+ } catch (UserException | UnsupportedEncodingException e) {
+ throw new LoadException(e.getMessage());
+ }
+
+ SparkLauncher launcher = new SparkLauncher();
+ // master | deployMode
+ // ------------|-------------
+ // yarn | cluster
+ // spark://xx | client
+ launcher.setMaster(resource.getMaster())
+ .setDeployMode(resource.getDeployMode().name().toLowerCase())
+ .setAppResource(appResourceHdfsPath)
+ .setMainClass(MAIN_CLASS)
+ .setAppName(String.format(ETL_JOB_NAME, loadLabel))
+ .addAppArgs(jobConfigHdfsPath);
+ // spark configs
+ for (Map.Entry<String, String> entry : resource.getSparkConfigs().entrySet()) {
+ launcher.setConf(entry.getKey(), entry.getValue());
+ }
+
+ // start app
+ SparkAppHandle handle = null;
+ State state = null;
+ String appId = null;
+ int retry = 0;
+ String errMsg = "start spark app failed. error: ";
+ try {
+ handle = launcher.startApplication(new SparkAppListener());
+ } 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: "
Review comment:
```suggestion
throw new LoadException(errMsg + " wait too much time for getting appi d. spark app state: "
```
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wyb commented on a change in pull request #3716: [Spark load][Fe 5/6] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
wyb commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r441988995
##########
File path: fe/src/main/java/org/apache/doris/common/Pair.java
##########
@@ -25,7 +27,9 @@
public class Pair<F, S> {
public static PairComparator<Pair<?, Comparable>> PAIR_VALUE_COMPARATOR = new PairComparator<>();
+ @SerializedName(value = "first")
Review comment:
I add a comment.
When using Pair for persistence, users need to guarantee that F and S can be serialized through Gson
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wyb commented on a change in pull request #3716: [Spark load][Fe 4/5] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
wyb commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r433352697
##########
File path: fe/src/main/java/org/apache/doris/load/loadv2/SparkEtlJobHandler.java
##########
@@ -0,0 +1,170 @@
+// 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.PaloFe;
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.catalog.SparkResource;
+import org.apache.doris.common.LoadException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.BrokerUtil;
+import org.apache.doris.load.loadv2.etl.EtlJobConfig;
+import org.apache.doris.thrift.TEtlState;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.spark.launcher.SparkAppHandle;
+import org.apache.spark.launcher.SparkAppHandle.Listener;
+import org.apache.spark.launcher.SparkAppHandle.State;
+import org.apache.spark.launcher.SparkLauncher;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * SparkEtlJobHandler is responsible for
+ * 1. submit spark etl job
+ * 2. get spark etl job status
+ * 3. kill spark etl job
+ * 4. get spark etl file paths
+ * 5. delete etl output path
+ */
+public class SparkEtlJobHandler {
+ private static final Logger LOG = LogManager.getLogger(SparkEtlJobHandler.class);
+
+ private static final String APP_RESOURCE_NAME = "palo-fe.jar";
+ private static final String CONFIG_FILE_NAME = "jobconfig.json";
+ private static final String APP_RESOURCE_LOCAL_PATH = PaloFe.DORIS_HOME_DIR + "/lib/" + APP_RESOURCE_NAME;
+ private static final String JOB_CONFIG_DIR = "configs";
+ private static final String MAIN_CLASS = "org.apache.doris.load.loadv2.etl.SparkEtlJob";
+ private static final String ETL_JOB_NAME = "doris__%s";
+ // 5min
+ private static final int GET_APPID_MAX_RETRY_TIMES = 300;
+ private static final int GET_APPID_SLEEP_MS = 1000;
+
+ class SparkAppListener implements Listener {
+ @Override
+ public void stateChanged(SparkAppHandle sparkAppHandle) {}
+
+ @Override
+ public void infoChanged(SparkAppHandle sparkAppHandle) {}
+ }
+
+ public void submitEtlJob(long loadJobId, String loadLabel, EtlJobConfig etlJobConfig, SparkResource resource,
+ BrokerDesc brokerDesc, SparkPendingTaskAttachment attachment) throws LoadException {
+ // delete outputPath
+ deleteEtlOutputPath(etlJobConfig.outputPath, brokerDesc);
+
+ // upload app resource and jobconfig to hdfs
+ String configsHdfsDir = etlJobConfig.outputPath + "/" + JOB_CONFIG_DIR + "/";
+ String appResourceHdfsPath = configsHdfsDir + APP_RESOURCE_NAME;
+ String jobConfigHdfsPath = configsHdfsDir + CONFIG_FILE_NAME;
+ try {
+ BrokerUtil.writeBrokerFile(APP_RESOURCE_LOCAL_PATH, appResourceHdfsPath, brokerDesc);
+ byte[] configData = etlJobConfig.configToJson().getBytes("UTF-8");
+ BrokerUtil.writeBrokerFile(configData, jobConfigHdfsPath, brokerDesc);
+ } catch (UserException | UnsupportedEncodingException e) {
+ throw new LoadException(e.getMessage());
+ }
+
+ SparkLauncher launcher = new SparkLauncher();
+ // master | deployMode
+ // ------------|-------------
+ // yarn | cluster
+ // spark://xx | client
+ launcher.setMaster(resource.getMaster())
+ .setDeployMode(resource.getDeployMode().name().toLowerCase())
+ .setAppResource(appResourceHdfsPath)
+ .setMainClass(MAIN_CLASS)
+ .setAppName(String.format(ETL_JOB_NAME, loadLabel))
+ .addAppArgs(jobConfigHdfsPath);
+ // spark configs
+ for (Map.Entry<String, String> entry : resource.getSparkConfigs().entrySet()) {
+ launcher.setConf(entry.getKey(), entry.getValue());
+ }
+
+ // start app
+ SparkAppHandle handle = null;
+ State state = null;
+ String appId = null;
+ int retry = 0;
+ String errMsg = "start spark app failed. error: ";
+ try {
+ handle = launcher.startApplication(new SparkAppListener());
+ } 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: "
Review comment:
errMsg already have a space at the end.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman merged pull request #3716: [Spark load][Fe 5/6] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
morningman merged pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] imay commented on a change in pull request #3716: [Spark load] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
imay commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r432214780
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +142,326 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ public static String readBrokerFile(String path, BrokerDesc brokerDesc) throws UserException {
Review comment:
prefer return byte[]
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +142,326 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ public static String readBrokerFile(String path, BrokerDesc brokerDesc) throws UserException {
+ TNetworkAddress address = getAddress(brokerDesc);
+ TPaloBrokerService.Client client = borrowClient(address);
+ boolean failed = true;
+ TBrokerFD fd = null;
+ try {
+ // get file size
+ TBrokerListPathRequest request = new TBrokerListPathRequest(
+ TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
+ TBrokerListResponse tBrokerListResponse = null;
+ try {
+ tBrokerListResponse = client.listPath(request);
+ } catch (TException e) {
+ reopenClient(client);
+ tBrokerListResponse = client.listPath(request);
+ }
+ if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
+ + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
+ }
+ List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
+ if (fileStatuses.size() != 1) {
+ throw new UserException("Broker files num error. path=" + path + ", broker=" + address
+ + ", files num: " + fileStatuses.size());
+ }
+
+ Preconditions.checkState(!fileStatuses.get(0).isIsDir());
+ long fileSize = fileStatuses.get(0).getSize();
+
+ // open reader
+ String clientId = FrontendOptions.getLocalHostAddress() + ":" + Config.rpc_port;
+ TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
+ TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
+ TBrokerOpenReaderResponse tOpenReaderResponse = null;
+ try {
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ }
+ if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
+ + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
+ }
+ fd = tOpenReaderResponse.getFd();
+
+ // read
+ TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
+ TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
+ TBrokerReadResponse tReadResponse = null;
+ try {
+ tReadResponse = client.pread(tPReadRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tReadResponse = client.pread(tPReadRequest);
+ }
+ if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker read failed. path=" + path + ", broker=" + address
+ + ", msg=" + tReadResponse.getOpStatus().getMessage());
+ }
+ failed = false;
+ byte[] data = tReadResponse.getData();
+ return new String(data, "UTF-8");
+ } catch (TException | UnsupportedEncodingException e) {
+ String failMsg = "Broker read file exception. path=" + path + ", broker=" + address;
+ LOG.warn(failMsg, e);
+ throw new UserException(failMsg);
+ } finally {
+ // close reader
+ if (fd != null) {
+ failed = true;
+ TBrokerCloseReaderRequest tCloseReaderRequest = new TBrokerCloseReaderRequest(
+ TBrokerVersion.VERSION_ONE, fd);
+ TBrokerOperationStatus tOperationStatus = null;
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException ex) {
+ LOG.warn("Broker close reader failed. path={}, address={}", path, address, ex);
+ }
+ }
+ if (tOperationStatus == null || tOperationStatus.getStatusCode() != TBrokerOperationStatusCode.OK) {
+ LOG.warn("Broker close reader failed. path={}, address={}, error={}", path, address,
+ tOperationStatus.getMessage());
+ } else {
+ failed = false;
+ }
+ }
+
+ // return client
+ returnClient(client, address, failed);
+ }
+ }
+
+ public static void writeBrokerFile(byte[] data, String destFilePath, BrokerDesc brokerDesc) throws UserException {
+ BrokerWriter writer = new BrokerWriter(destFilePath, brokerDesc);
+ try {
+ writer.open();
+ ByteBuffer byteBuffer = ByteBuffer.wrap(data);
+ writer.write(byteBuffer, data.length);
+ } finally {
+ writer.close();
+ }
+ }
+
+ public static void writeBrokerFile(String srcFilePath, String destFilePath,
Review comment:
```suggestion
public static void writeFile(String srcFilePath, String destFilePath,
```
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +142,326 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ public static String readBrokerFile(String path, BrokerDesc brokerDesc) throws UserException {
+ TNetworkAddress address = getAddress(brokerDesc);
+ TPaloBrokerService.Client client = borrowClient(address);
+ boolean failed = true;
+ TBrokerFD fd = null;
+ try {
+ // get file size
+ TBrokerListPathRequest request = new TBrokerListPathRequest(
+ TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
+ TBrokerListResponse tBrokerListResponse = null;
+ try {
+ tBrokerListResponse = client.listPath(request);
+ } catch (TException e) {
+ reopenClient(client);
+ tBrokerListResponse = client.listPath(request);
+ }
+ if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
+ + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
+ }
+ List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
+ if (fileStatuses.size() != 1) {
+ throw new UserException("Broker files num error. path=" + path + ", broker=" + address
+ + ", files num: " + fileStatuses.size());
+ }
+
+ Preconditions.checkState(!fileStatuses.get(0).isIsDir());
+ long fileSize = fileStatuses.get(0).getSize();
+
+ // open reader
+ String clientId = FrontendOptions.getLocalHostAddress() + ":" + Config.rpc_port;
+ TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
+ TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
+ TBrokerOpenReaderResponse tOpenReaderResponse = null;
+ try {
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ }
+ if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
+ + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
+ }
+ fd = tOpenReaderResponse.getFd();
+
+ // read
+ TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
+ TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
+ TBrokerReadResponse tReadResponse = null;
+ try {
+ tReadResponse = client.pread(tPReadRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tReadResponse = client.pread(tPReadRequest);
+ }
+ if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker read failed. path=" + path + ", broker=" + address
+ + ", msg=" + tReadResponse.getOpStatus().getMessage());
+ }
+ failed = false;
+ byte[] data = tReadResponse.getData();
+ return new String(data, "UTF-8");
+ } catch (TException | UnsupportedEncodingException e) {
+ String failMsg = "Broker read file exception. path=" + path + ", broker=" + address;
+ LOG.warn(failMsg, e);
+ throw new UserException(failMsg);
+ } finally {
+ // close reader
+ if (fd != null) {
+ failed = true;
+ TBrokerCloseReaderRequest tCloseReaderRequest = new TBrokerCloseReaderRequest(
+ TBrokerVersion.VERSION_ONE, fd);
+ TBrokerOperationStatus tOperationStatus = null;
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException ex) {
+ LOG.warn("Broker close reader failed. path={}, address={}", path, address, ex);
+ }
+ }
+ if (tOperationStatus == null || tOperationStatus.getStatusCode() != TBrokerOperationStatusCode.OK) {
+ LOG.warn("Broker close reader failed. path={}, address={}, error={}", path, address,
+ tOperationStatus.getMessage());
+ } else {
+ failed = false;
+ }
+ }
+
+ // return client
+ returnClient(client, address, failed);
+ }
+ }
+
+ public static void writeBrokerFile(byte[] data, String destFilePath, BrokerDesc brokerDesc) throws UserException {
Review comment:
should add comment for public function to help other to use it.
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +142,326 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ public static String readBrokerFile(String path, BrokerDesc brokerDesc) throws UserException {
+ TNetworkAddress address = getAddress(brokerDesc);
+ TPaloBrokerService.Client client = borrowClient(address);
+ boolean failed = true;
+ TBrokerFD fd = null;
+ try {
+ // get file size
+ TBrokerListPathRequest request = new TBrokerListPathRequest(
+ TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
+ TBrokerListResponse tBrokerListResponse = null;
+ try {
+ tBrokerListResponse = client.listPath(request);
+ } catch (TException e) {
+ reopenClient(client);
+ tBrokerListResponse = client.listPath(request);
+ }
+ if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
+ + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
+ }
+ List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
+ if (fileStatuses.size() != 1) {
+ throw new UserException("Broker files num error. path=" + path + ", broker=" + address
+ + ", files num: " + fileStatuses.size());
+ }
+
+ Preconditions.checkState(!fileStatuses.get(0).isIsDir());
+ long fileSize = fileStatuses.get(0).getSize();
+
+ // open reader
+ String clientId = FrontendOptions.getLocalHostAddress() + ":" + Config.rpc_port;
+ TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
+ TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
+ TBrokerOpenReaderResponse tOpenReaderResponse = null;
+ try {
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ }
+ if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
+ + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
+ }
+ fd = tOpenReaderResponse.getFd();
+
+ // read
+ TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
+ TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
+ TBrokerReadResponse tReadResponse = null;
+ try {
+ tReadResponse = client.pread(tPReadRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tReadResponse = client.pread(tPReadRequest);
+ }
+ if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker read failed. path=" + path + ", broker=" + address
+ + ", msg=" + tReadResponse.getOpStatus().getMessage());
+ }
+ failed = false;
+ byte[] data = tReadResponse.getData();
+ return new String(data, "UTF-8");
+ } catch (TException | UnsupportedEncodingException e) {
+ String failMsg = "Broker read file exception. path=" + path + ", broker=" + address;
+ LOG.warn(failMsg, e);
+ throw new UserException(failMsg);
+ } finally {
+ // close reader
+ if (fd != null) {
+ failed = true;
+ TBrokerCloseReaderRequest tCloseReaderRequest = new TBrokerCloseReaderRequest(
+ TBrokerVersion.VERSION_ONE, fd);
+ TBrokerOperationStatus tOperationStatus = null;
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ try {
+ tOperationStatus = client.closeReader(tCloseReaderRequest);
+ } catch (TException ex) {
+ LOG.warn("Broker close reader failed. path={}, address={}", path, address, ex);
+ }
+ }
+ if (tOperationStatus == null || tOperationStatus.getStatusCode() != TBrokerOperationStatusCode.OK) {
+ LOG.warn("Broker close reader failed. path={}, address={}, error={}", path, address,
+ tOperationStatus.getMessage());
+ } else {
+ failed = false;
+ }
+ }
+
+ // return client
+ returnClient(client, address, failed);
+ }
+ }
+
+ public static void writeBrokerFile(byte[] data, String destFilePath, BrokerDesc brokerDesc) throws UserException {
Review comment:
```suggestion
public static void writeFile(byte[] data, String destFilePath, BrokerDesc brokerDesc) throws UserException {
```
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wyb commented on a change in pull request #3716: [Spark load][Fe 4/5] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
wyb commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r433350943
##########
File path: fe/src/main/java/org/apache/doris/common/Pair.java
##########
@@ -25,7 +27,9 @@
public class Pair<F, S> {
public static PairComparator<Pair<?, Comparable>> PAIR_VALUE_COMPARATOR = new PairComparator<>();
+ @SerializedName(value = "first")
Review comment:
Users guarantee this when use Pair class？like Map and List.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] wyb commented on a change in pull request #3716: [Spark load][Fe 5/6] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
wyb commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r441988075
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +148,352 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ /**
+ * Read binary data from path with broker
+ * @param path
+ * @param brokerDesc
+ * @return byte[]
+ * @throws UserException if broker op failed or not only one file
+ */
+ public static byte[] readFile(String path, BrokerDesc brokerDesc) throws UserException {
+ TNetworkAddress address = getAddress(brokerDesc);
+ TPaloBrokerService.Client client = borrowClient(address);
+ boolean failed = true;
+ TBrokerFD fd = null;
+ try {
+ // get file size
+ TBrokerListPathRequest request = new TBrokerListPathRequest(
+ TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
+ TBrokerListResponse tBrokerListResponse = null;
+ try {
+ tBrokerListResponse = client.listPath(request);
+ } catch (TException e) {
+ reopenClient(client);
+ tBrokerListResponse = client.listPath(request);
+ }
+ if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
+ + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
+ }
+ List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
+ if (fileStatuses.size() != 1) {
+ throw new UserException("Broker files num error. path=" + path + ", broker=" + address
+ + ", files num: " + fileStatuses.size());
+ }
+
+ Preconditions.checkState(!fileStatuses.get(0).isIsDir());
+ long fileSize = fileStatuses.get(0).getSize();
+
+ // open reader
+ String clientId = FrontendOptions.getLocalHostAddress() + ":" + Config.rpc_port;
+ TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
+ TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
+ TBrokerOpenReaderResponse tOpenReaderResponse = null;
+ try {
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ }
+ if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
+ + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
+ }
+ fd = tOpenReaderResponse.getFd();
+
+ // read
+ TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
+ TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
+ TBrokerReadResponse tReadResponse = null;
+ try {
+ tReadResponse = client.pread(tPReadRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tReadResponse = client.pread(tPReadRequest);
+ }
+ if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker read failed. path=" + path + ", broker=" + address
+ + ", msg=" + tReadResponse.getOpStatus().getMessage());
+ }
+ failed = false;
+ return tReadResponse.getData();
Review comment:
ok
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [incubator-doris] morningman commented on a change in pull request #3716: [Spark load][Fe 5/6] Fe submit spark etl job
Posted by GitBox <gi...@apache.org>.
morningman commented on a change in pull request #3716:
URL: https://github.com/apache/incubator-doris/pull/3716#discussion_r437874451
##########
File path: fe/src/main/java/org/apache/doris/common/Pair.java
##########
@@ -25,7 +27,9 @@
public class Pair<F, S> {
public static PairComparator<Pair<?, Comparable>> PAIR_VALUE_COMPARATOR = new PairComparator<>();
+ @SerializedName(value = "first")
Review comment:
I checked, this is not work
##########
File path: fe/src/main/java/org/apache/doris/load/loadv2/SparkLoadJob.java
##########
@@ -127,6 +140,76 @@ private void setResourceInfo() throws DdlException {
brokerDesc = new BrokerDesc(sparkResource.getBroker(), brokerProperties);
}
+ @Override
+ public void beginTxn()
+ throws LabelAlreadyUsedException, BeginTransactionException, AnalysisException, DuplicatedRequestException {
+ transactionId = Catalog.getCurrentGlobalTransactionMgr()
+ .beginTransaction(dbId, Lists.newArrayList(fileGroupAggInfo.getAllTableIds()), label, null,
+ new TxnCoordinator(TxnSourceType.FE, FrontendOptions.getLocalHostAddress()),
+ LoadJobSourceType.FRONTEND, id, timeoutSecond);
+ }
+
+ @Override
+ protected void unprotectedExecuteJob() throws LoadException {
+ // create pending task
+ LoadTask task = new SparkLoadPendingTask(this, fileGroupAggInfo.getAggKeyToFileGroups(),
+ sparkResource, brokerDesc);
+ task.init();
+ idToTasks.put(task.getSignature(), task);
+ Catalog.getCurrentCatalog().getLoadTaskScheduler().submit(task);
+ }
+
+ @Override
+ public void onTaskFinished(TaskAttachment attachment) {
+ if (attachment instanceof SparkPendingTaskAttachment) {
+ onPendingTaskFinished((SparkPendingTaskAttachment) attachment);
+ }
+ }
+
+ private void onPendingTaskFinished(SparkPendingTaskAttachment attachment) {
+ writeLock();
+ try {
+ // check if job has been cancelled
+ if (isTxnDone()) {
+ LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
+ .add("state", state)
+ .add("error_msg", "this task will be ignored when job is: " + state)
+ .build());
+ return;
+ }
+
+ if (finishedTaskIds.contains(attachment.getTaskId())) {
+ LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
+ .add("task_id", attachment.getTaskId())
+ .add("error_msg", "this is a duplicated callback of pending task "
+ + "when broker already has loading task")
+ .build());
+ return;
+ }
+
+ // add task id into finishedTaskIds
+ finishedTaskIds.add(attachment.getTaskId());
+
+ sparkAppHandle = attachment.getHandle();
+ appId = attachment.getAppId();
+ etlOutputPath = attachment.getOutputPath();
+
+ executeEtl();
+ // log etl state
+ unprotectedLogUpdateStateInfo();
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ /**
+ * update etl start time and state in spark load job
+ */
+ private void executeEtl() {
+ etlStartTimestamp = System.currentTimeMillis();
+ state = JobState.ETL;
Review comment:
Print a log for tracing the state changing
##########
File path: fe/src/main/java/org/apache/doris/common/util/BrokerUtil.java
##########
@@ -139,4 +148,352 @@ public static String printBroker(String brokerName, TNetworkAddress address) {
return Lists.newArrayList(columns);
}
+ /**
+ * Read binary data from path with broker
+ * @param path
+ * @param brokerDesc
+ * @return byte[]
+ * @throws UserException if broker op failed or not only one file
+ */
+ public static byte[] readFile(String path, BrokerDesc brokerDesc) throws UserException {
+ TNetworkAddress address = getAddress(brokerDesc);
+ TPaloBrokerService.Client client = borrowClient(address);
+ boolean failed = true;
+ TBrokerFD fd = null;
+ try {
+ // get file size
+ TBrokerListPathRequest request = new TBrokerListPathRequest(
+ TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
+ TBrokerListResponse tBrokerListResponse = null;
+ try {
+ tBrokerListResponse = client.listPath(request);
+ } catch (TException e) {
+ reopenClient(client);
+ tBrokerListResponse = client.listPath(request);
+ }
+ if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
+ + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
+ }
+ List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
+ if (fileStatuses.size() != 1) {
+ throw new UserException("Broker files num error. path=" + path + ", broker=" + address
+ + ", files num: " + fileStatuses.size());
+ }
+
+ Preconditions.checkState(!fileStatuses.get(0).isIsDir());
+ long fileSize = fileStatuses.get(0).getSize();
+
+ // open reader
+ String clientId = FrontendOptions.getLocalHostAddress() + ":" + Config.rpc_port;
+ TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
+ TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
+ TBrokerOpenReaderResponse tOpenReaderResponse = null;
+ try {
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tOpenReaderResponse = client.openReader(tOpenReaderRequest);
+ }
+ if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
+ + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
+ }
+ fd = tOpenReaderResponse.getFd();
+
+ // read
+ TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
+ TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
+ TBrokerReadResponse tReadResponse = null;
+ try {
+ tReadResponse = client.pread(tPReadRequest);
+ } catch (TException e) {
+ reopenClient(client);
+ tReadResponse = client.pread(tPReadRequest);
+ }
+ if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
+ throw new UserException("Broker read failed. path=" + path + ", broker=" + address
+ + ", msg=" + tReadResponse.getOpStatus().getMessage());
+ }
+ failed = false;
+ return tReadResponse.getData();
Review comment:
broker's pread() method does not guarantee to read the specified length of data currently.
But #3881 is trying to solve this problem. Just for remind.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org