Skip to content

Commit

Permalink
[GOBBLIN-2046] Generalize gobblin-on-temporal ProcessWorkUnitImpl l…
Browse files Browse the repository at this point in the history 
…ogging not to presume `CopyEntity` (#3925)

Generalize gobblin-on-temporal `ProcessWorkUnitImpl` logging not to presume `CopyEntity`
  • Loading branch information
@phet
phet committed Apr 18, 2024
1 parent c8db780 commit b57ecbc
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 36 deletions.
80 changes: 44 additions & 36 deletions ...oral/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/ProcessWorkUnitImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -106,16 +106,12 @@ protected int execute(List<WorkUnit> workUnits, WorkUnitClaimCheck wu, JobState
GobblinMultiTaskAttempt.CommitPolicy multiTaskAttemptCommitPolicy = GobblinMultiTaskAttempt.CommitPolicy.IMMEDIATE; // as no speculative exec

SharedResourcesBroker<GobblinScopeTypes> resourcesBroker = JobStateUtils.getSharedResourcesBroker(jobState);
List<String> fileSourcePaths = workUnits.stream()
.map(workUnit -> getCopyableFileSourcePathDesc(workUnit, wu.getWorkUnitPath()))
.collect(Collectors.toList());
List<String> pathsToLog = getSourcePathsToLog(fileSourcePaths, jobState);
log.info("WU [{}] - submitting {} workUnits for copying source files: {}{}",
Optional<String> optWorkUnitsDesc = getOptWorkUnitsDesc(workUnits, wu.getWorkUnitPath(), jobState);
log.info("WU [{}] - submitting {} workUnits {}",
wu.getCorrelator(),
workUnits.size(),
pathsToLog.size() == workUnits.size() ? "" : ("**first " + pathsToLog.size() + " only** "),
pathsToLog);
log.debug("WU [{}] - (first) workUnit: {}", wu.getCorrelator(), workUnits.get(0).toJsonString());
optWorkUnitsDesc.orElse(""));
log.debug("WU [{}] - (first) workUnit: {}", wu.getCorrelator(), workUnits.isEmpty() ? "<<absent>>" : workUnits.get(0).toJsonString());

GobblinMultiTaskAttempt taskAttempt = GobblinMultiTaskAttempt.runWorkUnits(
jobState.getJobId(), containerId, jobState, workUnits,
Expand Down Expand Up @@ -154,46 +150,58 @@ public void onTaskCommitCompletion(Task task) {
};
}

protected String getCopyableFileSourcePathDesc(WorkUnit workUnit, String workUnitPath) {
return getOptFirstCopyableFile(Lists.newArrayList(workUnit), workUnitPath)
.map(copyableFile -> copyableFile.getOrigin().getPath().toString())
.orElse(
"<<not a CopyableFile("
+ getOptCopyEntityClass(workUnit, workUnitPath)
.map(Class::getSimpleName)
.orElse("<<not a CopyEntity!>>")
+ "): '" + workUnitPath + "'"
);
protected static Optional<String> getOptWorkUnitsDesc(List<WorkUnit> workUnits, String workUnitsPath, JobState jobState) {
List<String> fileSourcePaths = workUnits.stream()
.map(workUnit -> getOptCopyableFileSourcePathDesc(workUnit, workUnitsPath))
.filter(Optional::isPresent)
.map(Optional::get)
.collect(Collectors.toList());
if (fileSourcePaths.isEmpty()) {
// TODO - describe WorkUnits other than `CopyableFile`
return Optional.empty();
} else {
return Optional.of(getSourcePathsToLog(fileSourcePaths, jobState)).map(pathsToLog ->
"for copying source files: "
+ (pathsToLog.size() == workUnits.size() ? "" : ("**first " + pathsToLog.size() + " only** "))
+ pathsToLog
);
}
}

protected static Optional<String> getOptCopyableFileSourcePathDesc(WorkUnit workUnit, String workUnitPath) {
return getOptWorkUnitCopyEntityClass(workUnit, workUnitPath).flatMap(copyEntityClass ->
getOptCopyableFile(copyEntityClass, workUnit).map(copyableFile ->
copyableFile.getOrigin().getPath().toString()));
}

protected Optional<CopyableFile> getOptCopyableFile(TaskState taskState) {
return getOptCopyableFile(taskState, "taskState '" + taskState.getTaskId() + "'");
protected static Optional<CopyableFile> getOptCopyableFile(TaskState taskState) {
return getOptTaskStateCopyEntityClass(taskState).flatMap(copyEntityClass ->
getOptCopyableFile(copyEntityClass, taskState));
}

protected Optional<CopyableFile> getOptCopyableFile(State state, String logDesc) {
return getOptCopyEntityClass(state, logDesc).flatMap(copyEntityClass -> {
log.debug("(state) {} got (copyEntity) {}", state.getClass().getName(), copyEntityClass.getName());
if (CopyableFile.class.isAssignableFrom(copyEntityClass)) {
String serialization = state.getProp(CopySource.SERIALIZED_COPYABLE_FILE);
if (serialization != null) {
return Optional.of((CopyableFile) CopyEntity.deserialize(serialization));
}
protected static Optional<CopyableFile> getOptCopyableFile(Class<?> copyEntityClass, State state) {
log.debug("(state) {} got (copyEntity) {}", state.getClass().getName(), copyEntityClass.getName());
if (CopyableFile.class.isAssignableFrom(copyEntityClass)) {
String serialization = state.getProp(CopySource.SERIALIZED_COPYABLE_FILE);
if (serialization != null) {
return Optional.of((CopyableFile) CopyEntity.deserialize(serialization));
}
return Optional.empty();
});
}
return Optional.empty();
}

protected static Optional<Class<?>> getOptWorkUnitCopyEntityClass(WorkUnit workUnit, String workUnitPath) {
return getOptCopyEntityClass(workUnit, "workUnit '" + workUnitPath + "'");
}

protected Optional<CopyableFile> getOptFirstCopyableFile(List<WorkUnit> workUnits, String workUnitPath) {
return Optional.of(workUnits).filter(wus -> wus.size() > 0).flatMap(wus ->
getOptCopyableFile(wus.get(0), "workUnit '" + workUnitPath + "'")
);
protected static Optional<Class<?>> getOptTaskStateCopyEntityClass(TaskState taskState) {
return getOptCopyEntityClass(taskState, "taskState '" + taskState.getTaskId() + "'");
}

protected Optional<Class<?>> getOptCopyEntityClass(State state, String logDesc) {
protected static Optional<Class<?>> getOptCopyEntityClass(State state, String logDesc) {
try {
return Optional.of(CopySource.getCopyEntityClass(state));
} catch (IOException ioe) {
log.warn(logDesc + " - failed getting copy entity class:", ioe);
return Optional.empty();
}
}
Expand Down
7 changes: 7 additions & 0 deletions gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -647,33 +647,40 @@ private Map<String, LocalResource> addAppMasterLocalResources(ApplicationId appl
Map<String, LocalResource> appMasterResources = Maps.newHashMap();
FileSystem localFs = FileSystem.getLocal(new Configuration());

// NOTE: log after each step below for insight into what takes bulk of time
if (this.config.hasPath(GobblinYarnConfigurationKeys.LIB_JARS_DIR_KEY)) {
Path libJarsDestDir = new Path(appWorkDir, GobblinYarnConfigurationKeys.LIB_JARS_DIR_NAME);
addLibJars(new Path(this.config.getString(GobblinYarnConfigurationKeys.LIB_JARS_DIR_KEY)),
Optional.of(appMasterResources), libJarsDestDir, localFs);
LOGGER.info("Added lib jars to directory: {}", libJarsDestDir.toString());
}
if (this.config.hasPath(GobblinYarnConfigurationKeys.APP_MASTER_JARS_KEY)) {
Path appJarsDestDir = new Path(appMasterWorkDir, GobblinYarnConfigurationKeys.APP_JARS_DIR_NAME);
addAppJars(this.config.getString(GobblinYarnConfigurationKeys.APP_MASTER_JARS_KEY),
Optional.of(appMasterResources), appJarsDestDir, localFs);
LOGGER.info("Added app jars to directory: {}", appJarsDestDir.toString());
}
if (this.config.hasPath(GobblinYarnConfigurationKeys.APP_MASTER_FILES_LOCAL_KEY)) {
Path appFilesDestDir = new Path(appMasterWorkDir, GobblinYarnConfigurationKeys.APP_FILES_DIR_NAME);
addAppLocalFiles(this.config.getString(GobblinYarnConfigurationKeys.APP_MASTER_FILES_LOCAL_KEY),
Optional.of(appMasterResources), appFilesDestDir, localFs);
LOGGER.info("Added app local files to directory: {}", appFilesDestDir.toString());
}
if (this.config.hasPath(GobblinYarnConfigurationKeys.APP_MASTER_FILES_REMOTE_KEY)) {
YarnHelixUtils.addRemoteFilesToLocalResources(this.config.getString(GobblinYarnConfigurationKeys.APP_MASTER_FILES_REMOTE_KEY),
appMasterResources, yarnConfiguration);
LOGGER.info("Added remote files to local resources");
}
if (this.config.hasPath(GobblinYarnConfigurationKeys.APP_MASTER_ZIPS_REMOTE_KEY)) {
YarnHelixUtils.addRemoteZipsToLocalResources(this.config.getString(GobblinYarnConfigurationKeys.APP_MASTER_ZIPS_REMOTE_KEY),
appMasterResources, yarnConfiguration);
LOGGER.info("Added remote zips to local resources");
}
if (this.config.hasPath(GobblinClusterConfigurationKeys.JOB_CONF_PATH_KEY)) {
Path appFilesDestDir = new Path(appMasterWorkDir, GobblinYarnConfigurationKeys.APP_FILES_DIR_NAME);
addJobConfPackage(this.config.getString(GobblinClusterConfigurationKeys.JOB_CONF_PATH_KEY), appFilesDestDir,
appMasterResources);
LOGGER.info("Added job conf package to directory: {}", appFilesDestDir.toString());
}

return appMasterResources;
Expand Down

0 comments on commit b57ecbc

Please sign in to comment.