You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by qh...@apache.org on 2015/06/18 05:56:40 UTC
incubator-kylin git commit: KYLIN-838
Repository: incubator-kylin
Updated Branches:
refs/heads/0.8 aa238a89d -> 29d1d02e9
KYLIN-838
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/29d1d02e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/29d1d02e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/29d1d02e
Branch: refs/heads/0.8
Commit: 29d1d02e9bc0b306e0fd51fc91fef5aa18d95401
Parents: aa238a8
Author: qianhao.zhou <qi...@ebay.com>
Authored: Wed Jun 17 18:41:28 2015 +0800
Committer: qianhao.zhou <qi...@ebay.com>
Committed: Thu Jun 18 11:55:17 2015 +0800
----------------------------------------------------------------------
.../common/persistence/FileResourceStore.java | 35 +++++++++--
.../common/persistence/HBaseResourceStore.java | 63 +++++++++++++++-----
.../kylin/common/persistence/RawResource.java | 16 +++++
.../kylin/common/persistence/ResourceStore.java | 23 +++++++
.../org/apache/kylin/cube/CubeManagerTest.java | 15 +++++
.../org/apache/kylin/job/dao/ExecutableDao.java | 25 ++++----
.../kylin/job/execution/AbstractExecutable.java | 40 +++++++++----
.../kylin/job/manager/ExecutableManager.java | 57 +++++++++++-------
.../apache/kylin/rest/service/BasicService.java | 16 ++---
.../apache/kylin/rest/service/JobService.java | 54 ++++++++++++-----
10 files changed, 254 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java b/common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
index 33f5841..38c6d9d 100644
--- a/common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
+++ b/common/src/main/java/org/apache/kylin/common/persistence/FileResourceStore.java
@@ -18,16 +18,14 @@
package org.apache.kylin.common.persistence;
-import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
+import java.io.*;
import java.util.ArrayList;
+import java.util.List;
+import com.google.common.collect.Lists;
import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
public class FileResourceStore extends ResourceStore {
@@ -63,6 +61,31 @@ public class FileResourceStore extends ResourceStore {
}
@Override
+ protected List<RawResource> getAllResources(String rangeStart, String rangeEnd) throws IOException {
+ List<RawResource> result = Lists.newArrayList();
+ try {
+ String commonPrefix = StringUtils.getCommonPrefix(rangeEnd, rangeStart);
+ commonPrefix = commonPrefix.substring(0, commonPrefix.lastIndexOf("/") + 1);
+ final ArrayList<String> resources = listResourcesImpl(commonPrefix);
+ for (String resource : resources) {
+ if (resource.compareTo(rangeStart) >= 0 && resource.compareTo(rangeEnd) <= 0) {
+ if (existsImpl(resource)) {
+ result.add(new RawResource(getResourceImpl(resource), getResourceTimestampImpl(resource)));
+ }
+ }
+ }
+ return result;
+ } catch (IOException ex) {
+ for (RawResource rawResource : result) {
+ IOUtils.closeQuietly(rawResource.resource);
+ }
+ throw ex;
+ } catch (Exception ex) {
+ throw new UnsupportedOperationException(ex);
+ }
+ }
+
+ @Override
protected InputStream getResourceImpl(String resPath) throws IOException {
File f = file(resPath);
if (f.exists() && f.isFile())
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/common/src/main/java/org/apache/kylin/common/persistence/HBaseResourceStore.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/persistence/HBaseResourceStore.java b/common/src/main/java/org/apache/kylin/common/persistence/HBaseResourceStore.java
index e665298..f7014d9 100644
--- a/common/src/main/java/org/apache/kylin/common/persistence/HBaseResourceStore.java
+++ b/common/src/main/java/org/apache/kylin/common/persistence/HBaseResourceStore.java
@@ -18,6 +18,7 @@
package org.apache.kylin.common.persistence;
+import com.google.common.collect.Lists;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -30,14 +31,8 @@ import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.BytesUtil;
import org.apache.kylin.common.util.HadoopUtil;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.LinkedHashMap;
-import java.util.Map;
+import java.io.*;
+import java.util.*;
public class HBaseResourceStore extends ResourceStore {
@@ -137,11 +132,36 @@ public class HBaseResourceStore extends ResourceStore {
}
@Override
- protected InputStream getResourceImpl(String resPath) throws IOException {
- Result r = getByScan(resPath, B_FAMILY, B_COLUMN);
- if (r == null)
- return null;
+ protected List<RawResource> getAllResources(String rangeStart, String rangeEnd) throws IOException {
+ byte[] startRow = Bytes.toBytes(rangeStart);
+ byte[] endRow = Bytes.toBytes(rangeEnd);
+
+ Scan scan = new Scan(startRow, endRow);
+ scan.addColumn(B_FAMILY, B_COLUMN_TS);
+ scan.addColumn(B_FAMILY, B_COLUMN);
+
+ HTableInterface table = getConnection().getTable(getAllInOneTableName());
+ List<RawResource> result = Lists.newArrayList();
+ try {
+ ResultScanner scanner = table.getScanner(scan);
+ for (Result r : scanner) {
+ result.add(new RawResource(getInputStream(Bytes.toString(r.getRow()), r), getTimestamp(r)));
+ }
+ } catch (IOException e) {
+ for (RawResource rawResource : result) {
+ IOUtils.closeQuietly(rawResource.resource);
+ }
+ throw e;
+ } finally {
+ IOUtils.closeQuietly(table);
+ }
+ return result;
+ }
+ private InputStream getInputStream(String resPath, Result r) throws IOException {
+ if (r == null) {
+ return null;
+ }
byte[] value = r.getValue(B_FAMILY, B_COLUMN);
if (value.length == 0) {
Path redirectPath = bigCellHDFSPath(resPath);
@@ -154,13 +174,24 @@ public class HBaseResourceStore extends ResourceStore {
}
}
+ private long getTimestamp(Result r) {
+ if (r == null) {
+ return 0;
+ } else {
+ return Bytes.toLong(r.getValue(B_FAMILY, B_COLUMN_TS));
+ }
+ }
+
+ @Override
+ protected InputStream getResourceImpl(String resPath) throws IOException {
+ Result r = getByScan(resPath, B_FAMILY, B_COLUMN);
+ return getInputStream(resPath, r);
+ }
+
@Override
protected long getResourceTimestampImpl(String resPath) throws IOException {
Result r = getByScan(resPath, B_FAMILY, B_COLUMN_TS);
- if (r == null)
- return 0;
- else
- return Bytes.toLong(r.getValue(B_FAMILY, B_COLUMN_TS));
+ return getTimestamp(r);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/common/src/main/java/org/apache/kylin/common/persistence/RawResource.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/persistence/RawResource.java b/common/src/main/java/org/apache/kylin/common/persistence/RawResource.java
new file mode 100644
index 0000000..2c3238f
--- /dev/null
+++ b/common/src/main/java/org/apache/kylin/common/persistence/RawResource.java
@@ -0,0 +1,16 @@
+package org.apache.kylin.common.persistence;
+
+import java.io.InputStream;
+
+/**
+ */
+public class RawResource {
+
+ public final InputStream resource;
+ public final long timestamp;
+
+ public RawResource(InputStream resource, long timestamp) {
+ this.resource = resource;
+ this.timestamp = timestamp;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java b/common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
index a99ca5d..926b729 100644
--- a/common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
+++ b/common/src/main/java/org/apache/kylin/common/persistence/ResourceStore.java
@@ -24,6 +24,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -149,6 +150,28 @@ abstract public class ResourceStore {
return getResourceImpl(norm(resPath));
}
+ final public <T extends RootPersistentEntity> List<T> getAllResources(String rangeStart, String rangeEnd, Class<T> clazz, Serializer<T> serializer) throws IOException {
+ final List<RawResource> allResources = getAllResources(rangeStart, rangeEnd);
+ if (allResources.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<T> result = Lists.newArrayList();
+ try {
+ for (RawResource rawResource : allResources) {
+ final T element = serializer.deserialize(new DataInputStream(rawResource.resource));
+ element.setLastModified(rawResource.timestamp);
+ result.add(element);
+ }
+ return result;
+ } finally {
+ for (RawResource rawResource : allResources) {
+ IOUtils.closeQuietly(rawResource.resource);
+ }
+ }
+ }
+
+ abstract protected List<RawResource> getAllResources(String rangeStart, String rangeEnd) throws IOException;
+
abstract protected InputStream getResourceImpl(String resPath) throws IOException;
final public long getResourceTimestamp(String resPath) throws IOException {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
----------------------------------------------------------------------
diff --git a/cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java b/cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
index ddf89d9..64c4ea5 100644
--- a/cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
+++ b/cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
@@ -30,6 +30,10 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
import static org.junit.Assert.*;
/**
@@ -117,6 +121,17 @@ public class CubeManagerTest extends LocalFileMetadataTestCase {
}
+ @Test
+ public void testGetAllCubes() throws Exception {
+ final ResourceStore store = ResourceStore.getStore(getTestConfig());
+ final ArrayList<String> cubePath = store.listResources(ResourceStore.CUBE_RESOURCE_ROOT);
+ assertTrue(cubePath.size() > 1);
+ Collections.sort(cubePath);
+ final List<CubeInstance> cubes = store.getAllResources(cubePath.get(0), cubePath.get(cubePath.size() - 1), CubeInstance.class, CubeManager.CUBE_SERIALIZER);
+ assertEquals(cubePath.size(), cubes.size());
+ assertEquals(cubePath.size() - 1, store.getAllResources(cubePath.get(1), cubePath.get(cubePath.size() - 1), CubeInstance.class, CubeManager.CUBE_SERIALIZER).size());
+ }
+
@Test
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java b/job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
index d4983e2..8f81973 100644
--- a/job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
+++ b/job/src/main/java/org/apache/kylin/job/dao/ExecutableDao.java
@@ -19,8 +19,10 @@
package org.apache.kylin.job.dao;
import com.google.common.collect.Lists;
+import org.apache.commons.io.IOUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.persistence.JsonSerializer;
+import org.apache.kylin.common.persistence.RawResource;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.Serializer;
import org.apache.kylin.job.exception.PersistentException;
@@ -28,6 +30,7 @@ import org.apache.kylin.metadata.MetadataManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.DataInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -98,11 +101,10 @@ public class ExecutableDao {
if (resources == null) {
return Collections.emptyList();
}
- ArrayList<ExecutableOutputPO> result = new ArrayList<ExecutableOutputPO>(resources.size());
- for (String path : resources) {
- result.add(readJobOutputResource(path));
- }
- return result;
+ Collections.sort(resources);
+ String rangeStart = resources.get(0);
+ String rangeEnd = resources.get(resources.size() - 1);
+ return store.getAllResources(rangeStart, rangeEnd, ExecutableOutputPO.class, JOB_OUTPUT_SERIALIZER);
} catch (IOException e) {
logger.error("error get all Jobs:", e);
throw new PersistentException(e);
@@ -111,15 +113,14 @@ public class ExecutableDao {
public List<ExecutablePO> getJobs() throws PersistentException {
try {
- ArrayList<String> resources = store.listResources(JOB_PATH_ROOT);
- if (resources == null) {
+ final List<String> jobIds = store.listResources(JOB_PATH_ROOT);
+ if (jobIds.isEmpty()) {
return Collections.emptyList();
}
- ArrayList<ExecutablePO> result = new ArrayList<ExecutablePO>(resources.size());
- for (String path : resources) {
- result.add(readJobResource(path));
- }
- return result;
+ Collections.sort(jobIds);
+ String rangeStart = jobIds.get(0);
+ String rangeEnd = jobIds.get(jobIds.size() - 1);
+ return store.getAllResources(rangeStart, rangeEnd, ExecutablePO.class, JOB_SERIALIZER);
} catch (IOException e) {
logger.error("error get all Jobs:", e);
throw new PersistentException(e);
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java b/job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
index 54f05bf..d5b801b 100644
--- a/job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
+++ b/job/src/main/java/org/apache/kylin/job/execution/AbstractExecutable.java
@@ -22,14 +22,14 @@ import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.LogTitlePrinter;
import org.apache.kylin.common.util.MailService;
import org.apache.kylin.job.exception.ExecuteException;
import org.apache.kylin.job.impl.threadpool.DefaultContext;
import org.apache.kylin.job.manager.ExecutableManager;
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -231,7 +231,30 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
}
protected long getExtraInfoAsLong(String key, long defaultValue) {
- final String str = executableManager.getOutput(getId()).getExtra().get(key);
+ return getExtraInfoAsLong(executableManager.getOutput(getId()), key, defaultValue);
+ }
+
+ public static long getStartTime(Output output) {
+ return getExtraInfoAsLong(output, START_TIME, 0L);
+ }
+
+ public static long getEndTime(Output output) {
+ return getExtraInfoAsLong(output, END_TIME, 0L);
+ }
+
+ public static long getDuration(long startTime, long endTime) {
+ if (startTime == 0) {
+ return 0;
+ }
+ if (endTime == 0) {
+ return System.currentTimeMillis() - startTime;
+ } else {
+ return endTime - startTime;
+ }
+ }
+
+ public static long getExtraInfoAsLong(Output output, String key, long defaultValue) {
+ final String str = output.getExtra().get(key);
if (str != null) {
return Long.parseLong(str);
} else {
@@ -260,16 +283,7 @@ public abstract class AbstractExecutable implements Executable, Idempotent {
}
public final long getDuration() {
- final long startTime = getStartTime();
- if (startTime == 0) {
- return 0;
- }
- final long endTime = getEndTime();
- if (endTime == 0) {
- return System.currentTimeMillis() - startTime;
- } else {
- return endTime - startTime;
- }
+ return getDuration(getStartTime(), getEndTime());
}
/*
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
----------------------------------------------------------------------
diff --git a/job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java b/job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
index 3c066cb..a2b310b 100644
--- a/job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
+++ b/job/src/main/java/org/apache/kylin/job/manager/ExecutableManager.java
@@ -18,13 +18,10 @@
package org.apache.kylin.job.manager;
-import java.lang.reflect.Constructor;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import javax.annotation.Nullable;
-
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.util.ClassUtil;
import org.apache.kylin.job.dao.ExecutableDao;
@@ -32,18 +29,16 @@ import org.apache.kylin.job.dao.ExecutableOutputPO;
import org.apache.kylin.job.dao.ExecutablePO;
import org.apache.kylin.job.exception.IllegalStateTranferException;
import org.apache.kylin.job.exception.PersistentException;
-import org.apache.kylin.job.execution.AbstractExecutable;
-import org.apache.kylin.job.execution.DefaultChainedExecutable;
-import org.apache.kylin.job.execution.DefaultOutput;
-import org.apache.kylin.job.execution.ExecutableState;
-import org.apache.kylin.job.execution.Output;
+import org.apache.kylin.job.execution.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
+import javax.annotation.Nullable;
+import java.lang.reflect.Constructor;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
/**
*/
@@ -119,18 +114,36 @@ public class ExecutableManager {
try {
final ExecutableOutputPO jobOutput = executableDao.getJobOutput(uuid);
Preconditions.checkArgument(jobOutput != null, "there is no related output for job id:" + uuid);
- final DefaultOutput result = new DefaultOutput();
- result.setExtra(jobOutput.getInfo());
- result.setState(ExecutableState.valueOf(jobOutput.getStatus()));
- result.setVerboseMsg(jobOutput.getContent());
- result.setLastModified(jobOutput.getLastModified());
- return result;
+ return parseOutput(jobOutput);
} catch (PersistentException e) {
logger.error("fail to get job output:" + uuid, e);
throw new RuntimeException(e);
}
}
+ private DefaultOutput parseOutput(ExecutableOutputPO jobOutput) {
+ final DefaultOutput result = new DefaultOutput();
+ result.setExtra(jobOutput.getInfo());
+ result.setState(ExecutableState.valueOf(jobOutput.getStatus()));
+ result.setVerboseMsg(jobOutput.getContent());
+ result.setLastModified(jobOutput.getLastModified());
+ return result;
+ }
+
+ public Map<String, Output> getAllOutputs() {
+ try {
+ final List<ExecutableOutputPO> jobOutputs = executableDao.getJobOutputs();
+ HashMap<String, Output> result = Maps.newHashMap();
+ for (ExecutableOutputPO jobOutput : jobOutputs) {
+ result.put(jobOutput.getId(), parseOutput(jobOutput));
+ }
+ return result;
+ } catch (PersistentException e) {
+ logger.error("fail to get all job output:", e);
+ throw new RuntimeException(e);
+ }
+ }
+
public List<AbstractExecutable> getAllExecutables() {
try {
return Lists.transform(executableDao.getJobs(), new Function<ExecutablePO, AbstractExecutable>() {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/server/src/main/java/org/apache/kylin/rest/service/BasicService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/BasicService.java b/server/src/main/java/org/apache/kylin/rest/service/BasicService.java
index 0a8c808..a025535 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/BasicService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/BasicService.java
@@ -33,6 +33,7 @@ import org.apache.kylin.invertedindex.IIManager;
import org.apache.kylin.job.cube.CubingJob;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.execution.Output;
import org.apache.kylin.job.manager.ExecutableManager;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.project.ProjectInstance;
@@ -54,10 +55,7 @@ import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
-import java.util.EnumSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -176,7 +174,7 @@ public abstract class BasicService {
return IIManager.getInstance(getConfig());
}
- protected List<CubingJob> listAllCubingJobs(final String cubeName, final String projectName, final Set<ExecutableState> statusList) {
+ protected List<CubingJob> listAllCubingJobs(final String cubeName, final String projectName, final Set<ExecutableState> statusList, final Map<String, Output> allOutputs) {
List<CubingJob> results = Lists.newArrayList(FluentIterable.from(getExecutableManager().getAllExecutables()).filter(new Predicate<AbstractExecutable>() {
@Override
public boolean apply(AbstractExecutable executable) {
@@ -207,14 +205,18 @@ public abstract class BasicService {
}).filter(new Predicate<CubingJob>() {
@Override
public boolean apply(CubingJob executable) {
- return statusList.contains(executable.getStatus());
+ return statusList.contains(allOutputs.get(executable.getId()).getState());
}
}));
return results;
}
+ protected List<CubingJob> listAllCubingJobs(final String cubeName, final String projectName, final Set<ExecutableState> statusList) {
+ return listAllCubingJobs(cubeName, projectName, statusList, getExecutableManager().getAllOutputs());
+ }
+
protected List<CubingJob> listAllCubingJobs(final String cubeName, final String projectName) {
- return listAllCubingJobs(cubeName, projectName, EnumSet.allOf(ExecutableState.class));
+ return listAllCubingJobs(cubeName, projectName, EnumSet.allOf(ExecutableState.class), getExecutableManager().getAllOutputs());
}
protected static void close(ResultSet resultSet, Statement stat, Connection conn) {
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/29d1d02e/server/src/main/java/org/apache/kylin/rest/service/JobService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/JobService.java b/server/src/main/java/org/apache/kylin/rest/service/JobService.java
index 84b0943..b671383 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/JobService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/JobService.java
@@ -24,8 +24,8 @@ import com.google.common.collect.FluentIterable;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeBuilder;
+import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.model.CubeBuildTypeEnum;
import org.apache.kylin.job.JobInstance;
@@ -91,10 +91,11 @@ public class JobService extends BasicService {
states.add(parseToExecutableState(status));
}
}
- return Lists.newArrayList(FluentIterable.from(listAllCubingJobs(cubeName, projectName, states)).transform(new Function<CubingJob, JobInstance>() {
+ final Map<String, Output> allOutputs = getExecutableManager().getAllOutputs();
+ return Lists.newArrayList(FluentIterable.from(listAllCubingJobs(cubeName, projectName, states, allOutputs)).transform(new Function<CubingJob, JobInstance>() {
@Override
public JobInstance apply(CubingJob cubingJob) {
- return parseToJobInstance(cubingJob);
+ return parseToJobInstance(cubingJob, allOutputs);
}
}));
}
@@ -150,18 +151,18 @@ public class JobService extends BasicService {
throw new JobException("invalid build type:" + buildType);
}
getExecutableManager().addJob(job);
- return parseToJobInstance(job);
+ return getSingleJobInstance(job);
}
public JobInstance getJobInstance(String uuid) throws IOException, JobException {
- return parseToJobInstance(getExecutableManager().getJob(uuid));
+ return getSingleJobInstance(getExecutableManager().getJob(uuid));
}
public Output getOutput(String id) {
return getExecutableManager().getOutput(id);
}
- private JobInstance parseToJobInstance(AbstractExecutable job) {
+ private JobInstance getSingleJobInstance(AbstractExecutable job) {
if (job == null) {
return null;
}
@@ -180,31 +181,56 @@ public class JobService extends BasicService {
result.setDuration(cubeJob.getDuration() / 1000);
for (int i = 0; i < cubeJob.getTasks().size(); ++i) {
AbstractExecutable task = cubeJob.getTasks().get(i);
- result.addStep(parseToJobStep(task, i));
+ result.addStep(parseToJobStep(task, i, getExecutableManager().getOutput(task.getId())));
+ }
+ return result;
+ }
+
+ private JobInstance parseToJobInstance(AbstractExecutable job, Map<String, Output> outputs) {
+ if (job == null) {
+ return null;
+ }
+ Preconditions.checkState(job instanceof CubingJob, "illegal job type, id:" + job.getId());
+ CubingJob cubeJob = (CubingJob) job;
+ Output output = outputs.get(job.getId());
+ final JobInstance result = new JobInstance();
+ result.setName(job.getName());
+ result.setRelatedCube(cubeJob.getCubeName());
+ result.setRelatedSegment(cubeJob.getSegmentId());
+ result.setLastModified(output.getLastModified());
+ result.setSubmitter(cubeJob.getSubmitter());
+ result.setUuid(cubeJob.getId());
+ result.setType(CubeBuildTypeEnum.BUILD);
+ result.setStatus(parseToJobStatus(output.getState()));
+ result.setMrWaiting(AbstractExecutable.getExtraInfoAsLong(output, CubingJob.MAP_REDUCE_WAIT_TIME, 0L) / 1000);
+ result.setDuration(AbstractExecutable.getDuration(AbstractExecutable.getStartTime(output), AbstractExecutable.getEndTime(output)) / 1000);
+ for (int i = 0; i < cubeJob.getTasks().size(); ++i) {
+ AbstractExecutable task = cubeJob.getTasks().get(i);
+ result.addStep(parseToJobStep(task, i, outputs.get(task.getId())));
}
return result;
}
- private JobInstance.JobStep parseToJobStep(AbstractExecutable task, int i) {
+ private JobInstance.JobStep parseToJobStep(AbstractExecutable task, int i, Output stepOutput) {
+ Preconditions.checkNotNull(stepOutput);
JobInstance.JobStep result = new JobInstance.JobStep();
result.setId(task.getId());
result.setName(task.getName());
result.setSequenceID(i);
- result.setStatus(parseToJobStepStatus(task.getStatus()));
- final Output output = getExecutableManager().getOutput(task.getId());
- for (Map.Entry<String, String> entry : output.getExtra().entrySet()) {
+ result.setStatus(parseToJobStepStatus(stepOutput.getState()));
+ for (Map.Entry<String, String> entry : stepOutput.getExtra().entrySet()) {
if (entry.getKey() != null && entry.getValue() != null) {
result.putInfo(entry.getKey(), entry.getValue());
}
}
- result.setExecStartTime(task.getStartTime());
- result.setExecEndTime(task.getEndTime());
+ result.setExecStartTime(AbstractExecutable.getStartTime(stepOutput));
+ result.setExecEndTime(AbstractExecutable.getEndTime(stepOutput));
if (task instanceof ShellExecutable) {
result.setExecCmd(((ShellExecutable) task).getCmd());
}
if (task instanceof MapReduceExecutable) {
result.setExecCmd(((MapReduceExecutable) task).getMapReduceParams());
- result.setExecWaitTime(((MapReduceExecutable) task).getMapReduceWaitTime() / 1000);
+ result.setExecWaitTime(AbstractExecutable.getExtraInfoAsLong(stepOutput, MapReduceExecutable.MAP_REDUCE_WAIT_TIME, 0L) / 1000);
}
if (task instanceof HadoopShellExecutable) {
result.setExecCmd(((HadoopShellExecutable) task).getJobParams());