Skip to content

Feature/support tmp table cleanup #100

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
idleyui wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Feature/support tmp table cleanup #100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f91d230
feature: support tmp table cleanup
idleyui Feb 15, 2022
6b17b18
feature: fix clean in after resetjob & fix clean in catalogjob
idleyui Feb 21, 2022
9825e86
Merge branch 'master' into feature/support_tmp_table_cleanup
idleyui Feb 21, 2022
59735a9
feature: support verify & fix clean
idleyui Mar 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion mma-server/src/main/java/com/aliyun/odps/mma/server/job/McToOssCatalogJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
public class McToOssCatalogJob extends CatalogJob {

private static final Logger LOG = LogManager.getLogger(McToOssCatalogJob.class);
private List<Job> subJobs;

public McToOssCatalogJob(
Job parentJob,
Expand All @@ -49,7 +50,10 @@ public McToOssCatalogJob(

@Override
public synchronized List<Task> getExecutableTasks() {
List<Job> subJobs = getSubJobs();
subJobs = getSubJobs();
if (subJobs.isEmpty()) {
setStatusInternal(JobStatus.SUCCEEDED);
}
List<Task> ret = new LinkedList<>();

// Firstly, execute table jobs
Expand Down Expand Up @@ -86,4 +90,15 @@ void getJobsByObjectType(List<Task> ret, List<Job> subJobs, ObjectType objectTyp
ret.addAll(job.getExecutableTasks());
}
}

@Override
public boolean clean() {
// all subjobs is clean => true
// else false
boolean clean = true;
for (Job job : subJobs) {
clean = job.clean() && clean;
}
return clean;
}
}
7 changes: 7 additions & 0 deletions mma-server/src/main/java/com/aliyun/odps/mma/server/job/McToOssTableJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,4 +290,11 @@ public boolean clean() {
cleaned = true;
return false;
}

@Override
public synchronized boolean reset(boolean force) throws Exception {
cleaned = false;
return super.reset(force);
}

}
6 changes: 6 additions & 0 deletions mma-server/src/main/java/com/aliyun/odps/mma/server/job/OssToMcTableJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,4 +306,10 @@ public boolean clean() {
cleaned = true;
return false;
}

@Override
public synchronized boolean reset(boolean force) throws Exception {
cleaned = false;
return super.reset(force);
}
}
2 changes: 2 additions & 0 deletions mma-server/src/main/java/com/aliyun/odps/mma/server/meta/MetaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,7 @@ public List<Job> listJobs() {
public List<Job> listSubJobs(String parentJobId) {
try (SqlSession session = sqlSessionFactory.openSession(true)) {
JobDao dao = session.getMapper(JobDao.class);
dao.createSubJobTableIfNotExists(parentJobId);
return dao.selectSubJobs(parentJobId);
}
}
Expand All @@ -194,6 +195,7 @@ public List<Job> listJobsByStatus(JobStatus jobStatus) {
public List<Job> listSubJobsByStatus(String parentJobId, JobStatus jobStatus) {
try (SqlSession session = sqlSessionFactory.openSession(true)) {
JobDao dao = session.getMapper(JobDao.class);
dao.createSubJobTableIfNotExists(parentJobId);
return dao.selectSubJobsByJobStatus(parentJobId, jobStatus.name());
}
}
Expand Down
39 changes: 38 additions & 1 deletion ...erver/src/main/java/com/aliyun/odps/mma/server/task/McToOssTableDataTransmissionTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import com.aliyun.odps.mma.config.JobConfiguration;
import com.aliyun.odps.mma.server.action.ActionExecutionContext;
import com.aliyun.odps.mma.server.action.McToMcTableDataTransmissionAction;
import com.aliyun.odps.mma.server.action.McVerificationAction;
import com.aliyun.odps.mma.server.action.VerificationAction;
import com.aliyun.odps.mma.server.job.Job;
import com.aliyun.odps.mma.meta.MetaSource.TableMetaModel;

Expand Down Expand Up @@ -54,7 +56,42 @@ private void init() {
this,
context);
dag.addVertex(action);
// TODO: verification

McVerificationAction mcVerificationAction = new McVerificationAction(
id + ".McDataVerification",
config.get(JobConfiguration.DATA_SOURCE_MC_ACCESS_KEY_ID),
config.get(JobConfiguration.DATA_SOURCE_MC_ACCESS_KEY_SECRET),
executionProject,
config.get(JobConfiguration.DATA_SOURCE_MC_ENDPOINT),
source,
true,
this,
context);
dag.addVertex(mcVerificationAction);

McVerificationAction ossVerificationAction = new McVerificationAction(
id + ".OssDataVerification",
config.get(JobConfiguration.DATA_SOURCE_MC_ACCESS_KEY_ID),
config.get(JobConfiguration.DATA_SOURCE_MC_ACCESS_KEY_SECRET),
executionProject,
config.get(JobConfiguration.DATA_SOURCE_MC_ENDPOINT),
dest,
false,
this,
context);
dag.addVertex(ossVerificationAction);

VerificationAction verificationAction = new VerificationAction(
id + ".FinalVerification",
source,
this,
context);
dag.addVertex(verificationAction);

dag.addEdge(action, mcVerificationAction);
dag.addEdge(action, ossVerificationAction);
dag.addEdge(ossVerificationAction, verificationAction);
dag.addEdge(mcVerificationAction, verificationAction);
}

@Override
Expand Down
38 changes: 38 additions & 0 deletions ...erver/src/main/java/com/aliyun/odps/mma/server/task/OssToMcTableDataTransmissionTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
import com.aliyun.odps.mma.config.JobConfiguration;
import com.aliyun.odps.mma.server.action.ActionExecutionContext;
import com.aliyun.odps.mma.server.action.McToMcTableDataTransmissionAction;
import com.aliyun.odps.mma.server.action.McVerificationAction;
import com.aliyun.odps.mma.server.action.VerificationAction;
import com.aliyun.odps.mma.server.job.Job;
import com.aliyun.odps.mma.meta.MetaSource.TableMetaModel;

Expand Down Expand Up @@ -54,6 +56,42 @@ private void init() {
this,
context);
dag.addVertex(action);

McVerificationAction mcVerificationAction = new McVerificationAction(
id + ".McDataVerification",
config.get(JobConfiguration.DATA_DEST_MC_ACCESS_KEY_ID),
config.get(JobConfiguration.DATA_DEST_MC_ACCESS_KEY_SECRET),
executionProject,
config.get(JobConfiguration.DATA_DEST_MC_ENDPOINT),
source,
true,
this,
context);
dag.addVertex(mcVerificationAction);

McVerificationAction ossVerificationAction = new McVerificationAction(
id + ".OssDataVerification",
config.get(JobConfiguration.DATA_DEST_MC_ACCESS_KEY_ID),
config.get(JobConfiguration.DATA_DEST_MC_ACCESS_KEY_SECRET),
executionProject,
config.get(JobConfiguration.DATA_DEST_MC_ENDPOINT),
dest,
false,
this,
context);
dag.addVertex(ossVerificationAction);

VerificationAction verificationAction = new VerificationAction(
id + ".FinalVerification",
source,
this,
context);
dag.addVertex(verificationAction);

dag.addEdge(action, mcVerificationAction);
dag.addEdge(action, ossVerificationAction);
dag.addEdge(ossVerificationAction, verificationAction);
dag.addEdge(mcVerificationAction, verificationAction);
}

@Override
Expand Down