You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2015/03/03 14:08:34 UTC
[18/31] incubator-ignite git commit: # IGNITE-386: WIP on internal
namings (3).
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Counter.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Counter.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Counter.java
deleted file mode 100644
index c4fc599..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Counter.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.mapreduce.*;
-import org.apache.ignite.internal.processors.hadoop.counter.*;
-
-import java.io.*;
-
-/**
- * Adapter from own counter implementation into Hadoop API Counter od version 2.0.
- */
-public class GridHadoopV2Counter implements Counter {
- /** Delegate. */
- private final HadoopLongCounter cntr;
-
- /**
- * Creates new instance with given delegate.
- *
- * @param cntr Internal counter.
- */
- public GridHadoopV2Counter(HadoopLongCounter cntr) {
- assert cntr != null : "counter must be non-null";
-
- this.cntr = cntr;
- }
-
- /** {@inheritDoc} */
- @Override public void setDisplayName(String displayName) {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public String getName() {
- return cntr.name();
- }
-
- /** {@inheritDoc} */
- @Override public String getDisplayName() {
- return getName();
- }
-
- /** {@inheritDoc} */
- @Override public long getValue() {
- return cntr.value();
- }
-
- /** {@inheritDoc} */
- @Override public void setValue(long val) {
- cntr.value(val);
- }
-
- /** {@inheritDoc} */
- @Override public void increment(long incr) {
- cntr.increment(incr);
- }
-
- /** {@inheritDoc} */
- @Override public Counter getUnderlyingCounter() {
- return this;
- }
-
- /** {@inheritDoc} */
- @Override public void write(DataOutput out) throws IOException {
- throw new UnsupportedOperationException("not implemented");
- }
-
- /** {@inheritDoc} */
- @Override public void readFields(DataInput in) throws IOException {
- throw new UnsupportedOperationException("not implemented");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Job.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Job.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Job.java
deleted file mode 100644
index 5145bdb..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Job.java
+++ /dev/null
@@ -1,280 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.split.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.processors.hadoop.fs.*;
-import org.apache.ignite.internal.processors.hadoop.v1.*;
-import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jdk8.backport.*;
-
-import java.io.*;
-import java.lang.reflect.*;
-import java.util.*;
-import java.util.Queue;
-import java.util.concurrent.*;
-
-import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.*;
-
-/**
- * Hadoop job implementation for v2 API.
- */
-public class GridHadoopV2Job implements GridHadoopJob {
- /** */
- private final JobConf jobConf;
-
- /** */
- private final JobContextImpl jobCtx;
-
- /** Hadoop job ID. */
- private final GridHadoopJobId jobId;
-
- /** Job info. */
- protected GridHadoopJobInfo jobInfo;
-
- /** */
- private final JobID hadoopJobID;
-
- /** */
- private final GridHadoopV2JobResourceManager rsrcMgr;
-
- /** */
- private final ConcurrentMap<T2<GridHadoopTaskType, Integer>, GridFutureAdapter<GridHadoopTaskContext>> ctxs =
- new ConcurrentHashMap8<>();
-
- /** Pooling task context class and thus class loading environment. */
- private final Queue<Class<?>> taskCtxClsPool = new ConcurrentLinkedQueue<>();
-
- /** Local node ID */
- private UUID locNodeId;
-
- /** Serialized JobConf. */
- private volatile byte[] jobConfData;
-
- /**
- * @param jobId Job ID.
- * @param jobInfo Job info.
- * @param log Logger.
- */
- public GridHadoopV2Job(GridHadoopJobId jobId, final HadoopDefaultJobInfo jobInfo, IgniteLogger log) {
- assert jobId != null;
- assert jobInfo != null;
-
- this.jobId = jobId;
- this.jobInfo = jobInfo;
-
- hadoopJobID = new JobID(jobId.globalId().toString(), jobId.localId());
-
- HadoopClassLoader clsLdr = (HadoopClassLoader)getClass().getClassLoader();
-
- // Before create JobConf instance we should set new context class loader.
- Thread.currentThread().setContextClassLoader(clsLdr);
-
- jobConf = new JobConf();
-
- HadoopFileSystemsUtils.setupFileSystems(jobConf);
-
- Thread.currentThread().setContextClassLoader(null);
-
- for (Map.Entry<String,String> e : jobInfo.properties().entrySet())
- jobConf.set(e.getKey(), e.getValue());
-
- jobCtx = new JobContextImpl(jobConf, hadoopJobID);
-
- rsrcMgr = new GridHadoopV2JobResourceManager(jobId, jobCtx, log);
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopJobId id() {
- return jobId;
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopJobInfo info() {
- return jobInfo;
- }
-
- /** {@inheritDoc} */
- @Override public Collection<GridHadoopInputSplit> input() throws IgniteCheckedException {
- Thread.currentThread().setContextClassLoader(jobConf.getClassLoader());
-
- try {
- String jobDirPath = jobConf.get(MRJobConfig.MAPREDUCE_JOB_DIR);
-
- if (jobDirPath == null) { // Probably job was submitted not by hadoop client.
- // Assume that we have needed classes and try to generate input splits ourself.
- if (jobConf.getUseNewMapper())
- return GridHadoopV2Splitter.splitJob(jobCtx);
- else
- return GridHadoopV1Splitter.splitJob(jobConf);
- }
-
- Path jobDir = new Path(jobDirPath);
-
- try (FileSystem fs = FileSystem.get(jobDir.toUri(), jobConf)) {
- JobSplit.TaskSplitMetaInfo[] metaInfos = SplitMetaInfoReader.readSplitMetaInfo(hadoopJobID, fs, jobConf,
- jobDir);
-
- if (F.isEmpty(metaInfos))
- throw new IgniteCheckedException("No input splits found.");
-
- Path splitsFile = JobSubmissionFiles.getJobSplitFile(jobDir);
-
- try (FSDataInputStream in = fs.open(splitsFile)) {
- Collection<GridHadoopInputSplit> res = new ArrayList<>(metaInfos.length);
-
- for (JobSplit.TaskSplitMetaInfo metaInfo : metaInfos) {
- long off = metaInfo.getStartOffset();
-
- String[] hosts = metaInfo.getLocations();
-
- in.seek(off);
-
- String clsName = Text.readString(in);
-
- GridHadoopFileBlock block = GridHadoopV1Splitter.readFileBlock(clsName, in, hosts);
-
- if (block == null)
- block = GridHadoopV2Splitter.readFileBlock(clsName, in, hosts);
-
- res.add(block != null ? block : new GridHadoopExternalSplit(hosts, off));
- }
-
- return res;
- }
- }
- catch (Throwable e) {
- throw transformException(e);
- }
- }
- finally {
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopTaskContext getTaskContext(GridHadoopTaskInfo info) throws IgniteCheckedException {
- T2<GridHadoopTaskType, Integer> locTaskId = new T2<>(info.type(), info.taskNumber());
-
- GridFutureAdapter<GridHadoopTaskContext> fut = ctxs.get(locTaskId);
-
- if (fut != null)
- return fut.get();
-
- GridFutureAdapter<GridHadoopTaskContext> old = ctxs.putIfAbsent(locTaskId, fut = new GridFutureAdapter<>());
-
- if (old != null)
- return old.get();
-
- Class<?> cls = taskCtxClsPool.poll();
-
- try {
- if (cls == null) {
- // If there is no pooled class, then load new one.
- HadoopClassLoader ldr = new HadoopClassLoader(rsrcMgr.classPath());
-
- cls = ldr.loadClass(GridHadoopV2TaskContext.class.getName());
- }
-
- Constructor<?> ctr = cls.getConstructor(GridHadoopTaskInfo.class, GridHadoopJob.class,
- GridHadoopJobId.class, UUID.class, DataInput.class);
-
- if (jobConfData == null)
- synchronized(jobConf) {
- if (jobConfData == null) {
- ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
- jobConf.write(new DataOutputStream(buf));
-
- jobConfData = buf.toByteArray();
- }
- }
-
- GridHadoopTaskContext res = (GridHadoopTaskContext)ctr.newInstance(info, this, jobId, locNodeId,
- new DataInputStream(new ByteArrayInputStream(jobConfData)));
-
- fut.onDone(res);
-
- return res;
- }
- catch (Throwable e) {
- IgniteCheckedException te = transformException(e);
-
- fut.onDone(te);
-
- throw te;
- }
- }
-
- /** {@inheritDoc} */
- @Override public void initialize(boolean external, UUID locNodeId) throws IgniteCheckedException {
- this.locNodeId = locNodeId;
-
- Thread.currentThread().setContextClassLoader(jobConf.getClassLoader());
-
- try {
- rsrcMgr.prepareJobEnvironment(!external, jobLocalDir(locNodeId, jobId));
- }
- finally {
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void dispose(boolean external) throws IgniteCheckedException {
- if (rsrcMgr != null && !external) {
- File jobLocDir = jobLocalDir(locNodeId, jobId);
-
- if (jobLocDir.exists())
- U.delete(jobLocDir);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void prepareTaskEnvironment(GridHadoopTaskInfo info) throws IgniteCheckedException {
- rsrcMgr.prepareTaskWorkDir(taskLocalDir(locNodeId, info));
- }
-
- /** {@inheritDoc} */
- @Override public void cleanupTaskEnvironment(GridHadoopTaskInfo info) throws IgniteCheckedException {
- GridHadoopTaskContext ctx = ctxs.remove(new T2<>(info.type(), info.taskNumber())).get();
-
- taskCtxClsPool.offer(ctx.getClass());
-
- File locDir = taskLocalDir(locNodeId, info);
-
- if (locDir.exists())
- U.delete(locDir);
- }
-
- /** {@inheritDoc} */
- @Override public void cleanupStagingDirectory() {
- if (rsrcMgr != null)
- rsrcMgr.cleanupStagingDirectory();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2JobResourceManager.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2JobResourceManager.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2JobResourceManager.java
deleted file mode 100644
index fdda3f6..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2JobResourceManager.java
+++ /dev/null
@@ -1,305 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.processors.hadoop.fs.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.net.*;
-import java.nio.file.*;
-import java.util.*;
-
-/**
- * Provides all resources are needed to the job execution. Downloads the main jar, the configuration and additional
- * files are needed to be placed on local files system.
- */
-public class GridHadoopV2JobResourceManager {
- /** Hadoop job context. */
- private final JobContextImpl ctx;
-
- /** Logger. */
- private final IgniteLogger log;
-
- /** Job ID. */
- private final GridHadoopJobId jobId;
-
- /** Class path list. */
- private URL[] clsPath;
-
- /** Set of local resources. */
- private final Collection<File> rsrcSet = new HashSet<>();
-
- /** Staging directory to delivery job jar and config to the work nodes. */
- private Path stagingDir;
-
- /**
- * Creates new instance.
- * @param jobId Job ID.
- * @param ctx Hadoop job context.
- * @param log Logger.
- */
- public GridHadoopV2JobResourceManager(GridHadoopJobId jobId, JobContextImpl ctx, IgniteLogger log) {
- this.jobId = jobId;
- this.ctx = ctx;
- this.log = log.getLogger(GridHadoopV2JobResourceManager.class);
- }
-
- /**
- * Set working directory in local file system.
- *
- * @param dir Working directory.
- * @throws IOException If fails.
- */
- private void setLocalFSWorkingDirectory(File dir) throws IOException {
- JobConf cfg = ctx.getJobConf();
-
- Thread.currentThread().setContextClassLoader(cfg.getClassLoader());
-
- try {
- cfg.set(HadoopFileSystemsUtils.LOC_FS_WORK_DIR_PROP, dir.getAbsolutePath());
-
- if(!cfg.getBoolean("fs.file.impl.disable.cache", false))
- FileSystem.getLocal(cfg).setWorkingDirectory(new Path(dir.getAbsolutePath()));
- }
- finally {
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /**
- * Prepare job resources. Resolve the classpath list and download it if needed.
- *
- * @param download {@code true} If need to download resources.
- * @param jobLocDir Work directory for the job.
- * @throws IgniteCheckedException If failed.
- */
- public void prepareJobEnvironment(boolean download, File jobLocDir) throws IgniteCheckedException {
- try {
- if (jobLocDir.exists())
- throw new IgniteCheckedException("Local job directory already exists: " + jobLocDir.getAbsolutePath());
-
- JobConf cfg = ctx.getJobConf();
-
- String mrDir = cfg.get("mapreduce.job.dir");
-
- if (mrDir != null) {
- stagingDir = new Path(new URI(mrDir));
-
- if (download) {
- FileSystem fs = FileSystem.get(stagingDir.toUri(), cfg);
-
- if (!fs.exists(stagingDir))
- throw new IgniteCheckedException("Failed to find map-reduce submission directory (does not exist): " +
- stagingDir);
-
- if (!FileUtil.copy(fs, stagingDir, jobLocDir, false, cfg))
- throw new IgniteCheckedException("Failed to copy job submission directory contents to local file system " +
- "[path=" + stagingDir + ", locDir=" + jobLocDir.getAbsolutePath() + ", jobId=" + jobId + ']');
- }
-
- File jarJobFile = new File(jobLocDir, "job.jar");
-
- Collection<URL> clsPathUrls = new ArrayList<>();
-
- clsPathUrls.add(jarJobFile.toURI().toURL());
-
- rsrcSet.add(jarJobFile);
- rsrcSet.add(new File(jobLocDir, "job.xml"));
-
- processFiles(jobLocDir, ctx.getCacheFiles(), download, false, null, MRJobConfig.CACHE_LOCALFILES);
- processFiles(jobLocDir, ctx.getCacheArchives(), download, true, null, MRJobConfig.CACHE_LOCALARCHIVES);
- processFiles(jobLocDir, ctx.getFileClassPaths(), download, false, clsPathUrls, null);
- processFiles(jobLocDir, ctx.getArchiveClassPaths(), download, true, clsPathUrls, null);
-
- if (!clsPathUrls.isEmpty()) {
- clsPath = new URL[clsPathUrls.size()];
-
- clsPathUrls.toArray(clsPath);
- }
- }
- else if (!jobLocDir.mkdirs())
- throw new IgniteCheckedException("Failed to create local job directory: " + jobLocDir.getAbsolutePath());
-
- setLocalFSWorkingDirectory(jobLocDir);
- }
- catch (URISyntaxException | IOException e) {
- throw new IgniteCheckedException(e);
- }
- }
-
- /**
- * Process list of resources.
- *
- * @param jobLocDir Job working directory.
- * @param files Array of {@link java.net.URI} or {@link org.apache.hadoop.fs.Path} to process resources.
- * @param download {@code true}, if need to download. Process class path only else.
- * @param extract {@code true}, if need to extract archive.
- * @param clsPathUrls Collection to add resource as classpath resource.
- * @param rsrcNameProp Property for resource name array setting.
- * @throws IOException If failed.
- */
- private void processFiles(File jobLocDir, @Nullable Object[] files, boolean download, boolean extract,
- @Nullable Collection<URL> clsPathUrls, @Nullable String rsrcNameProp) throws IOException {
- if (F.isEmptyOrNulls(files))
- return;
-
- Collection<String> res = new ArrayList<>();
-
- for (Object pathObj : files) {
- String locName = null;
- Path srcPath;
-
- if (pathObj instanceof URI) {
- URI uri = (URI)pathObj;
-
- locName = uri.getFragment();
-
- srcPath = new Path(uri);
- }
- else
- srcPath = (Path)pathObj;
-
- if (locName == null)
- locName = srcPath.getName();
-
- File dstPath = new File(jobLocDir.getAbsolutePath(), locName);
-
- res.add(locName);
-
- rsrcSet.add(dstPath);
-
- if (clsPathUrls != null)
- clsPathUrls.add(dstPath.toURI().toURL());
-
- if (!download)
- continue;
-
- JobConf cfg = ctx.getJobConf();
-
- FileSystem dstFs = FileSystem.getLocal(cfg);
-
- FileSystem srcFs = srcPath.getFileSystem(cfg);
-
- if (extract) {
- File archivesPath = new File(jobLocDir.getAbsolutePath(), ".cached-archives");
-
- if (!archivesPath.exists() && !archivesPath.mkdir())
- throw new IOException("Failed to create directory " +
- "[path=" + archivesPath + ", jobId=" + jobId + ']');
-
- File archiveFile = new File(archivesPath, locName);
-
- FileUtil.copy(srcFs, srcPath, dstFs, new Path(archiveFile.toString()), false, cfg);
-
- String archiveNameLC = archiveFile.getName().toLowerCase();
-
- if (archiveNameLC.endsWith(".jar"))
- RunJar.unJar(archiveFile, dstPath);
- else if (archiveNameLC.endsWith(".zip"))
- FileUtil.unZip(archiveFile, dstPath);
- else if (archiveNameLC.endsWith(".tar.gz") ||
- archiveNameLC.endsWith(".tgz") ||
- archiveNameLC.endsWith(".tar"))
- FileUtil.unTar(archiveFile, dstPath);
- else
- throw new IOException("Cannot unpack archive [path=" + srcPath + ", jobId=" + jobId + ']');
- }
- else
- FileUtil.copy(srcFs, srcPath, dstFs, new Path(dstPath.toString()), false, cfg);
- }
-
- if (!res.isEmpty() && rsrcNameProp != null)
- ctx.getJobConf().setStrings(rsrcNameProp, res.toArray(new String[res.size()]));
- }
-
- /**
- * Prepares working directory for the task.
- *
- * <ul>
- * <li>Creates working directory.</li>
- * <li>Creates symbolic links to all job resources in working directory.</li>
- * </ul>
- *
- * @param path Path to working directory of the task.
- * @throws IgniteCheckedException If fails.
- */
- public void prepareTaskWorkDir(File path) throws IgniteCheckedException {
- try {
- if (path.exists())
- throw new IOException("Task local directory already exists: " + path);
-
- if (!path.mkdir())
- throw new IOException("Failed to create directory: " + path);
-
- for (File resource : rsrcSet) {
- File symLink = new File(path, resource.getName());
-
- try {
- Files.createSymbolicLink(symLink.toPath(), resource.toPath());
- }
- catch (IOException e) {
- String msg = "Unable to create symlink \"" + symLink + "\" to \"" + resource + "\".";
-
- if (U.isWindows() && e instanceof FileSystemException)
- msg += "\n\nAbility to create symbolic links is required!\n" +
- "On Windows platform you have to grant permission 'Create symbolic links'\n" +
- "to your user or run the Accelerator as Administrator.\n";
-
- throw new IOException(msg, e);
- }
- }
- }
- catch (IOException e) {
- throw new IgniteCheckedException("Unable to prepare local working directory for the task " +
- "[jobId=" + jobId + ", path=" + path+ ']', e);
- }
- }
-
- /**
- * Cleans up job staging directory.
- */
- public void cleanupStagingDirectory() {
- try {
- if (stagingDir != null)
- stagingDir.getFileSystem(ctx.getJobConf()).delete(stagingDir, true);
- }
- catch (Exception e) {
- log.error("Failed to remove job staging directory [path=" + stagingDir + ", jobId=" + jobId + ']' , e);
- }
- }
-
- /**
- * Returns array of class path for current job.
- *
- * @return Class path collection.
- */
- @Nullable public URL[] classPath() {
- return clsPath;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2MapTask.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2MapTask.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2MapTask.java
deleted file mode 100644
index be0bea2..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2MapTask.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-import org.apache.hadoop.mapreduce.lib.map.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.*;
-
-/**
- * Hadoop map task implementation for v2 API.
- */
-public class GridHadoopV2MapTask extends GridHadoopV2Task {
- /**
- * @param taskInfo Task info.
- */
- public GridHadoopV2MapTask(GridHadoopTaskInfo taskInfo) {
- super(taskInfo);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"ConstantConditions", "unchecked"})
- @Override public void run0(GridHadoopV2TaskContext taskCtx) throws IgniteCheckedException {
- GridHadoopInputSplit split = info().inputSplit();
-
- InputSplit nativeSplit;
-
- if (split instanceof GridHadoopFileBlock) {
- GridHadoopFileBlock block = (GridHadoopFileBlock)split;
-
- nativeSplit = new FileSplit(new Path(block.file().toString()), block.start(), block.length(), null);
- }
- else
- nativeSplit = (InputSplit)taskCtx.getNativeSplit(split);
-
- assert nativeSplit != null;
-
- OutputFormat outputFormat = null;
- Exception err = null;
-
- JobContextImpl jobCtx = taskCtx.jobContext();
-
- try {
- InputFormat inFormat = ReflectionUtils.newInstance(jobCtx.getInputFormatClass(),
- hadoopContext().getConfiguration());
-
- RecordReader reader = inFormat.createRecordReader(nativeSplit, hadoopContext());
-
- reader.initialize(nativeSplit, hadoopContext());
-
- hadoopContext().reader(reader);
-
- GridHadoopJobInfo jobInfo = taskCtx.job().info();
-
- outputFormat = jobInfo.hasCombiner() || jobInfo.hasReducer() ? null : prepareWriter(jobCtx);
-
- Mapper mapper = ReflectionUtils.newInstance(jobCtx.getMapperClass(), hadoopContext().getConfiguration());
-
- try {
- mapper.run(new WrappedMapper().getMapContext(hadoopContext()));
- }
- finally {
- closeWriter();
- }
-
- commit(outputFormat);
- }
- catch (InterruptedException e) {
- err = e;
-
- Thread.currentThread().interrupt();
-
- throw new IgniteInterruptedCheckedException(e);
- }
- catch (Exception e) {
- err = e;
-
- throw new IgniteCheckedException(e);
- }
- finally {
- if (err != null)
- abort(outputFormat);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Partitioner.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Partitioner.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Partitioner.java
deleted file mode 100644
index 0883520..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Partitioner.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.conf.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-
-/**
- * Hadoop partitioner adapter for v2 API.
- */
-public class GridHadoopV2Partitioner implements GridHadoopPartitioner {
- /** Partitioner instance. */
- private Partitioner<Object, Object> part;
-
- /**
- * @param cls Hadoop partitioner class.
- * @param conf Job configuration.
- */
- public GridHadoopV2Partitioner(Class<? extends Partitioner<?, ?>> cls, Configuration conf) {
- part = (Partitioner<Object, Object>) ReflectionUtils.newInstance(cls, conf);
- }
-
- /** {@inheritDoc} */
- @Override public int partition(Object key, Object val, int parts) {
- return part.getPartition(key, val, parts);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2ReduceTask.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2ReduceTask.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2ReduceTask.java
deleted file mode 100644
index 146e05c..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2ReduceTask.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.lib.reduce.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.*;
-
-/**
- * Hadoop reduce task implementation for v2 API.
- */
-public class GridHadoopV2ReduceTask extends GridHadoopV2Task {
- /** {@code True} if reduce, {@code false} if combine. */
- private final boolean reduce;
-
- /**
- * Constructor.
- *
- * @param taskInfo Task info.
- * @param reduce {@code True} if reduce, {@code false} if combine.
- */
- public GridHadoopV2ReduceTask(GridHadoopTaskInfo taskInfo, boolean reduce) {
- super(taskInfo);
-
- this.reduce = reduce;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings({"ConstantConditions", "unchecked"})
- @Override public void run0(GridHadoopV2TaskContext taskCtx) throws IgniteCheckedException {
- OutputFormat outputFormat = null;
- Exception err = null;
-
- JobContextImpl jobCtx = taskCtx.jobContext();
-
- try {
- outputFormat = reduce || !taskCtx.job().info().hasReducer() ? prepareWriter(jobCtx) : null;
-
- Reducer reducer = ReflectionUtils.newInstance(reduce ? jobCtx.getReducerClass() : jobCtx.getCombinerClass(),
- jobCtx.getConfiguration());
-
- try {
- reducer.run(new WrappedReducer().getReducerContext(hadoopContext()));
- }
- finally {
- closeWriter();
- }
-
- commit(outputFormat);
- }
- catch (InterruptedException e) {
- err = e;
-
- Thread.currentThread().interrupt();
-
- throw new IgniteInterruptedCheckedException(e);
- }
- catch (Exception e) {
- err = e;
-
- throw new IgniteCheckedException(e);
- }
- finally {
- if (err != null)
- abort(outputFormat);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2SetupTask.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2SetupTask.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2SetupTask.java
deleted file mode 100644
index 54eda25..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2SetupTask.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.OutputCommitter;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.*;
-
-import java.io.*;
-
-/**
- * Hadoop setup task (prepares job).
- */
-public class GridHadoopV2SetupTask extends GridHadoopV2Task {
- /**
- * Constructor.
- *
- * @param taskInfo task info.
- */
- public GridHadoopV2SetupTask(GridHadoopTaskInfo taskInfo) {
- super(taskInfo);
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("ConstantConditions")
- @Override protected void run0(GridHadoopV2TaskContext taskCtx) throws IgniteCheckedException {
- try {
- JobContextImpl jobCtx = taskCtx.jobContext();
-
- OutputFormat outputFormat = getOutputFormat(jobCtx);
-
- outputFormat.checkOutputSpecs(jobCtx);
-
- OutputCommitter committer = outputFormat.getOutputCommitter(hadoopContext());
-
- if (committer != null)
- committer.setupJob(jobCtx);
- }
- catch (ClassNotFoundException | IOException e) {
- throw new IgniteCheckedException(e);
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
-
- throw new IgniteInterruptedCheckedException(e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Splitter.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Splitter.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Splitter.java
deleted file mode 100644
index 68338a6..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Splitter.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.util.*;
-
-/**
- * Hadoop API v2 splitter.
- */
-public class GridHadoopV2Splitter {
- /** */
- private static final String[] EMPTY_HOSTS = {};
-
- /**
- * @param ctx Job context.
- * @return Collection of mapped splits.
- * @throws IgniteCheckedException If mapping failed.
- */
- public static Collection<GridHadoopInputSplit> splitJob(JobContext ctx) throws IgniteCheckedException {
- try {
- InputFormat<?, ?> format = ReflectionUtils.newInstance(ctx.getInputFormatClass(), ctx.getConfiguration());
-
- assert format != null;
-
- List<InputSplit> splits = format.getSplits(ctx);
-
- Collection<GridHadoopInputSplit> res = new ArrayList<>(splits.size());
-
- int id = 0;
-
- for (InputSplit nativeSplit : splits) {
- if (nativeSplit instanceof FileSplit) {
- FileSplit s = (FileSplit)nativeSplit;
-
- res.add(new GridHadoopFileBlock(s.getLocations(), s.getPath().toUri(), s.getStart(), s.getLength()));
- }
- else
- res.add(HadoopUtils.wrapSplit(id, nativeSplit, nativeSplit.getLocations()));
-
- id++;
- }
-
- return res;
- }
- catch (IOException | ClassNotFoundException e) {
- throw new IgniteCheckedException(e);
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
-
- throw new IgniteInterruptedCheckedException(e);
- }
- }
-
- /**
- * @param clsName Input split class name.
- * @param in Input stream.
- * @param hosts Optional hosts.
- * @return File block or {@code null} if it is not a {@link FileSplit} instance.
- * @throws IgniteCheckedException If failed.
- */
- public static GridHadoopFileBlock readFileBlock(String clsName, DataInput in, @Nullable String[] hosts)
- throws IgniteCheckedException {
- if (!FileSplit.class.getName().equals(clsName))
- return null;
-
- FileSplit split = new FileSplit();
-
- try {
- split.readFields(in);
- }
- catch (IOException e) {
- throw new IgniteCheckedException(e);
- }
-
- if (hosts == null)
- hosts = EMPTY_HOSTS;
-
- return new GridHadoopFileBlock(hosts, split.getPath().toUri(), split.getStart(), split.getLength());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Task.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Task.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Task.java
deleted file mode 100644
index 37697c6..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2Task.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.util.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-
-/**
- * Extended Hadoop v2 task.
- */
-public abstract class GridHadoopV2Task extends GridHadoopTask {
- /** Hadoop context. */
- private GridHadoopV2Context hadoopCtx;
-
- /**
- * Constructor.
- *
- * @param taskInfo Task info.
- */
- protected GridHadoopV2Task(GridHadoopTaskInfo taskInfo) {
- super(taskInfo);
- }
-
- /** {@inheritDoc} */
- @Override public void run(GridHadoopTaskContext taskCtx) throws IgniteCheckedException {
- GridHadoopV2TaskContext ctx = (GridHadoopV2TaskContext)taskCtx;
-
- hadoopCtx = new GridHadoopV2Context(ctx);
-
- run0(ctx);
- }
-
- /**
- * Internal task routine.
- *
- * @param taskCtx Task context.
- * @throws IgniteCheckedException
- */
- protected abstract void run0(GridHadoopV2TaskContext taskCtx) throws IgniteCheckedException;
-
- /**
- * @return hadoop context.
- */
- protected GridHadoopV2Context hadoopContext() {
- return hadoopCtx;
- }
-
- /**
- * Create and configure an OutputFormat instance.
- *
- * @param jobCtx Job context.
- * @return Instance of OutputFormat is specified in job configuration.
- * @throws ClassNotFoundException If specified class not found.
- */
- protected OutputFormat getOutputFormat(JobContext jobCtx) throws ClassNotFoundException {
- return ReflectionUtils.newInstance(jobCtx.getOutputFormatClass(), hadoopContext().getConfiguration());
- }
-
- /**
- * Put write into Hadoop context and return associated output format instance.
- *
- * @param jobCtx Job context.
- * @return Output format.
- * @throws IgniteCheckedException In case of Grid exception.
- * @throws InterruptedException In case of interrupt.
- */
- protected OutputFormat prepareWriter(JobContext jobCtx)
- throws IgniteCheckedException, InterruptedException {
- try {
- OutputFormat outputFormat = getOutputFormat(jobCtx);
-
- assert outputFormat != null;
-
- OutputCommitter outCommitter = outputFormat.getOutputCommitter(hadoopCtx);
-
- if (outCommitter != null)
- outCommitter.setupTask(hadoopCtx);
-
- RecordWriter writer = outputFormat.getRecordWriter(hadoopCtx);
-
- hadoopCtx.writer(writer);
-
- return outputFormat;
- }
- catch (IOException | ClassNotFoundException e) {
- throw new IgniteCheckedException(e);
- }
- }
-
- /**
- * Closes writer.
- *
- * @throws Exception If fails and logger hasn't been specified.
- */
- protected void closeWriter() throws Exception {
- RecordWriter writer = hadoopCtx.writer();
-
- if (writer != null)
- writer.close(hadoopCtx);
- }
-
- /**
- * Setup task.
- *
- * @param outputFormat Output format.
- * @throws IOException In case of IO exception.
- * @throws InterruptedException In case of interrupt.
- */
- protected void setup(@Nullable OutputFormat outputFormat) throws IOException, InterruptedException {
- if (hadoopCtx.writer() != null) {
- assert outputFormat != null;
-
- outputFormat.getOutputCommitter(hadoopCtx).setupTask(hadoopCtx);
- }
- }
-
- /**
- * Commit task.
- *
- * @param outputFormat Output format.
- * @throws IgniteCheckedException In case of Grid exception.
- * @throws IOException In case of IO exception.
- * @throws InterruptedException In case of interrupt.
- */
- protected void commit(@Nullable OutputFormat outputFormat) throws IgniteCheckedException, IOException, InterruptedException {
- if (hadoopCtx.writer() != null) {
- assert outputFormat != null;
-
- OutputCommitter outputCommitter = outputFormat.getOutputCommitter(hadoopCtx);
-
- if (outputCommitter.needsTaskCommit(hadoopCtx))
- outputCommitter.commitTask(hadoopCtx);
- }
- }
-
- /**
- * Abort task.
- *
- * @param outputFormat Output format.
- */
- protected void abort(@Nullable OutputFormat outputFormat) {
- if (hadoopCtx.writer() != null) {
- assert outputFormat != null;
-
- try {
- outputFormat.getOutputCommitter(hadoopCtx).abortTask(hadoopCtx);
- }
- catch (IOException ignore) {
- // Ignore.
- }
- catch (InterruptedException ignore) {
- Thread.currentThread().interrupt();
- }
- }
- }
-
- /** {@inheritDoc} */
- @Override public void cancel() {
- hadoopCtx.cancel();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2TaskContext.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2TaskContext.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2TaskContext.java
deleted file mode 100644
index 476d8f0..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopV2TaskContext.java
+++ /dev/null
@@ -1,443 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.conf.*;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.io.serializer.*;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapred.JobID;
-import org.apache.hadoop.mapred.TaskAttemptID;
-import org.apache.hadoop.mapred.TaskID;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.processors.hadoop.counter.*;
-import org.apache.ignite.internal.processors.hadoop.fs.*;
-import org.apache.ignite.internal.processors.hadoop.v1.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.util.*;
-
-import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.*;
-import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.*;
-
-/**
- * Context for task execution.
- */
-public class GridHadoopV2TaskContext extends GridHadoopTaskContext {
- /** */
- private static final boolean COMBINE_KEY_GROUPING_SUPPORTED;
-
- /**
- * Check for combiner grouping support (available since Hadoop 2.3).
- */
- static {
- boolean ok;
-
- try {
- JobContext.class.getDeclaredMethod("getCombinerKeyGroupingComparator");
-
- ok = true;
- }
- catch (NoSuchMethodException ignore) {
- ok = false;
- }
-
- COMBINE_KEY_GROUPING_SUPPORTED = ok;
- }
-
- /** Flag is set if new context-object code is used for running the mapper. */
- private final boolean useNewMapper;
-
- /** Flag is set if new context-object code is used for running the reducer. */
- private final boolean useNewReducer;
-
- /** Flag is set if new context-object code is used for running the combiner. */
- private final boolean useNewCombiner;
-
- /** */
- private final JobContextImpl jobCtx;
-
- /** Set if task is to cancelling. */
- private volatile boolean cancelled;
-
- /** Current task. */
- private volatile GridHadoopTask task;
-
- /** Local node ID */
- private UUID locNodeId;
-
- /** Counters for task. */
- private final GridHadoopCounters cntrs = new HadoopCountersImpl();
-
- /**
- * @param taskInfo Task info.
- * @param job Job.
- * @param jobId Job ID.
- * @param locNodeId Local node ID.
- * @param jobConfDataInput DataInput for read JobConf.
- */
- public GridHadoopV2TaskContext(GridHadoopTaskInfo taskInfo, GridHadoopJob job, GridHadoopJobId jobId,
- @Nullable UUID locNodeId, DataInput jobConfDataInput) throws IgniteCheckedException {
- super(taskInfo, job);
- this.locNodeId = locNodeId;
-
- // Before create JobConf instance we should set new context class loader.
- Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
-
- try {
- JobConf jobConf = new JobConf();
-
- try {
- jobConf.readFields(jobConfDataInput);
- }
- catch (IOException e) {
- throw new IgniteCheckedException(e);
- }
-
- // For map-reduce jobs prefer local writes.
- jobConf.setBooleanIfUnset(PARAM_IGFS_PREFER_LOCAL_WRITES, true);
-
- jobCtx = new JobContextImpl(jobConf, new JobID(jobId.globalId().toString(), jobId.localId()));
-
- useNewMapper = jobConf.getUseNewMapper();
- useNewReducer = jobConf.getUseNewReducer();
- useNewCombiner = jobConf.getCombinerClass() == null;
- }
- finally {
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /** {@inheritDoc} */
- @Override public <T extends GridHadoopCounter> T counter(String grp, String name, Class<T> cls) {
- return cntrs.counter(grp, name, cls);
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopCounters counters() {
- return cntrs;
- }
-
- /**
- * Creates appropriate task from current task info.
- *
- * @return Task.
- */
- private GridHadoopTask createTask() {
- boolean isAbort = taskInfo().type() == GridHadoopTaskType.ABORT;
-
- switch (taskInfo().type()) {
- case SETUP:
- return useNewMapper ? new GridHadoopV2SetupTask(taskInfo()) : new GridHadoopV1SetupTask(taskInfo());
-
- case MAP:
- return useNewMapper ? new GridHadoopV2MapTask(taskInfo()) : new GridHadoopV1MapTask(taskInfo());
-
- case REDUCE:
- return useNewReducer ? new GridHadoopV2ReduceTask(taskInfo(), true) :
- new GridHadoopV1ReduceTask(taskInfo(), true);
-
- case COMBINE:
- return useNewCombiner ? new GridHadoopV2ReduceTask(taskInfo(), false) :
- new GridHadoopV1ReduceTask(taskInfo(), false);
-
- case COMMIT:
- case ABORT:
- return useNewReducer ? new GridHadoopV2CleanupTask(taskInfo(), isAbort) :
- new GridHadoopV1CleanupTask(taskInfo(), isAbort);
-
- default:
- return null;
- }
- }
-
- /** {@inheritDoc} */
- @Override public void run() throws IgniteCheckedException {
- try {
- Thread.currentThread().setContextClassLoader(jobConf().getClassLoader());
-
- try {
- task = createTask();
- }
- catch (Throwable e) {
- throw transformException(e);
- }
-
- if (cancelled)
- throw new HadoopTaskCancelledException("Task cancelled.");
-
- try {
- task.run(this);
- }
- catch (Throwable e) {
- throw transformException(e);
- }
- }
- finally {
- task = null;
-
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void cancel() {
- cancelled = true;
-
- GridHadoopTask t = task;
-
- if (t != null)
- t.cancel();
- }
-
- /** {@inheritDoc} */
- @Override public void prepareTaskEnvironment() throws IgniteCheckedException {
- File locDir;
-
- switch(taskInfo().type()) {
- case MAP:
- case REDUCE:
- job().prepareTaskEnvironment(taskInfo());
-
- locDir = taskLocalDir(locNodeId, taskInfo());
-
- break;
-
- default:
- locDir = jobLocalDir(locNodeId, taskInfo().jobId());
- }
-
- Thread.currentThread().setContextClassLoader(jobConf().getClassLoader());
-
- try {
- FileSystem fs = FileSystem.get(jobConf());
-
- HadoopFileSystemsUtils.setUser(fs, jobConf().getUser());
-
- LocalFileSystem locFs = FileSystem.getLocal(jobConf());
-
- locFs.setWorkingDirectory(new Path(locDir.getAbsolutePath()));
- }
- catch (Throwable e) {
- throw transformException(e);
- }
- finally {
- Thread.currentThread().setContextClassLoader(null);
- }
- }
-
- /** {@inheritDoc} */
- @Override public void cleanupTaskEnvironment() throws IgniteCheckedException {
- job().cleanupTaskEnvironment(taskInfo());
- }
-
- /**
- * Creates Hadoop attempt ID.
- *
- * @return Attempt ID.
- */
- public TaskAttemptID attemptId() {
- TaskID tid = new TaskID(jobCtx.getJobID(), taskType(taskInfo().type()), taskInfo().taskNumber());
-
- return new TaskAttemptID(tid, taskInfo().attempt());
- }
-
- /**
- * @param type Task type.
- * @return Hadoop task type.
- */
- private TaskType taskType(GridHadoopTaskType type) {
- switch (type) {
- case SETUP:
- return TaskType.JOB_SETUP;
- case MAP:
- case COMBINE:
- return TaskType.MAP;
-
- case REDUCE:
- return TaskType.REDUCE;
-
- case COMMIT:
- case ABORT:
- return TaskType.JOB_CLEANUP;
-
- default:
- return null;
- }
- }
-
- /**
- * Gets job configuration of the task.
- *
- * @return Job configuration.
- */
- public JobConf jobConf() {
- return jobCtx.getJobConf();
- }
-
- /**
- * Gets job context of the task.
- *
- * @return Job context.
- */
- public JobContextImpl jobContext() {
- return jobCtx;
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopPartitioner partitioner() throws IgniteCheckedException {
- Class<?> partClsOld = jobConf().getClass("mapred.partitioner.class", null);
-
- if (partClsOld != null)
- return new GridHadoopV1Partitioner(jobConf().getPartitionerClass(), jobConf());
-
- try {
- return new GridHadoopV2Partitioner(jobCtx.getPartitionerClass(), jobConf());
- }
- catch (ClassNotFoundException e) {
- throw new IgniteCheckedException(e);
- }
- }
-
- /**
- * Gets serializer for specified class.
- *
- * @param cls Class.
- * @param jobConf Job configuration.
- * @return Appropriate serializer.
- */
- @SuppressWarnings("unchecked")
- private GridHadoopSerialization getSerialization(Class<?> cls, Configuration jobConf) throws IgniteCheckedException {
- A.notNull(cls, "cls");
-
- SerializationFactory factory = new SerializationFactory(jobConf);
-
- Serialization<?> serialization = factory.getSerialization(cls);
-
- if (serialization == null)
- throw new IgniteCheckedException("Failed to find serialization for: " + cls.getName());
-
- if (serialization.getClass() == WritableSerialization.class)
- return new GridHadoopWritableSerialization((Class<? extends Writable>)cls);
-
- return new GridHadoopSerializationWrapper(serialization, cls);
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopSerialization keySerialization() throws IgniteCheckedException {
- return getSerialization(jobCtx.getMapOutputKeyClass(), jobConf());
- }
-
- /** {@inheritDoc} */
- @Override public GridHadoopSerialization valueSerialization() throws IgniteCheckedException {
- return getSerialization(jobCtx.getMapOutputValueClass(), jobConf());
- }
-
- /** {@inheritDoc} */
- @Override public Comparator<Object> sortComparator() {
- return (Comparator<Object>)jobCtx.getSortComparator();
- }
-
- /** {@inheritDoc} */
- @Override public Comparator<Object> groupComparator() {
- Comparator<?> res;
-
- switch (taskInfo().type()) {
- case COMBINE:
- res = COMBINE_KEY_GROUPING_SUPPORTED ?
- jobContext().getCombinerKeyGroupingComparator() : jobContext().getGroupingComparator();
-
- break;
-
- case REDUCE:
- res = jobContext().getGroupingComparator();
-
- break;
-
- default:
- return null;
- }
-
- if (res != null && res.getClass() != sortComparator().getClass())
- return (Comparator<Object>)res;
-
- return null;
- }
-
- /**
- * @param split Split.
- * @return Native Hadoop split.
- * @throws IgniteCheckedException if failed.
- */
- @SuppressWarnings("unchecked")
- public Object getNativeSplit(GridHadoopInputSplit split) throws IgniteCheckedException {
- if (split instanceof GridHadoopExternalSplit)
- return readExternalSplit((GridHadoopExternalSplit)split);
-
- if (split instanceof GridHadoopSplitWrapper)
- return unwrapSplit((GridHadoopSplitWrapper)split);
-
- throw new IllegalStateException("Unknown split: " + split);
- }
-
- /**
- * @param split External split.
- * @return Native input split.
- * @throws IgniteCheckedException If failed.
- */
- @SuppressWarnings("unchecked")
- private Object readExternalSplit(GridHadoopExternalSplit split) throws IgniteCheckedException {
- Path jobDir = new Path(jobConf().get(MRJobConfig.MAPREDUCE_JOB_DIR));
-
- try (FileSystem fs = FileSystem.get(jobDir.toUri(), jobConf());
- FSDataInputStream in = fs.open(JobSubmissionFiles.getJobSplitFile(jobDir))) {
-
- in.seek(split.offset());
-
- String clsName = Text.readString(in);
-
- Class<?> cls = jobConf().getClassByName(clsName);
-
- assert cls != null;
-
- Serialization serialization = new SerializationFactory(jobConf()).getSerialization(cls);
-
- Deserializer deserializer = serialization.getDeserializer(cls);
-
- deserializer.open(in);
-
- Object res = deserializer.deserialize(null);
-
- deserializer.close();
-
- assert res != null;
-
- return res;
- }
- catch (IOException | ClassNotFoundException e) {
- throw new IgniteCheckedException(e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopWritableSerialization.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopWritableSerialization.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopWritableSerialization.java
deleted file mode 100644
index 4361ad4..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/GridHadoopWritableSerialization.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * 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.ignite.internal.processors.hadoop.v2;
-
-import org.apache.hadoop.io.*;
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.hadoop.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-
-/**
- * Optimized serialization for Hadoop {@link Writable} types.
- */
-public class GridHadoopWritableSerialization implements GridHadoopSerialization {
- /** */
- private final Class<? extends Writable> cls;
-
- /**
- * @param cls Class.
- */
- public GridHadoopWritableSerialization(Class<? extends Writable> cls) {
- assert cls != null;
-
- this.cls = cls;
- }
-
- /** {@inheritDoc} */
- @Override public void write(DataOutput out, Object obj) throws IgniteCheckedException {
- assert cls.isAssignableFrom(obj.getClass()) : cls + " " + obj.getClass();
-
- try {
- ((Writable)obj).write(out);
- }
- catch (IOException e) {
- throw new IgniteCheckedException(e);
- }
- }
-
- /** {@inheritDoc} */
- @Override public Object read(DataInput in, @Nullable Object obj) throws IgniteCheckedException {
- Writable w = obj == null ? U.newInstance(cls) : cls.cast(obj);
-
- try {
- w.readFields(in);
- }
- catch (IOException e) {
- throw new IgniteCheckedException(e);
- }
-
- return w;
- }
-
- /** {@inheritDoc} */
- @Override public void close() {
- // No-op.
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopExternalSplit.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopExternalSplit.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopExternalSplit.java
new file mode 100644
index 0000000..8e968b2
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopExternalSplit.java
@@ -0,0 +1,87 @@
+/*
+ * 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.ignite.internal.processors.hadoop.v2;
+
+import org.apache.ignite.internal.processors.hadoop.*;
+
+import java.io.*;
+
+/**
+ * Split serialized in external file.
+ */
+public class HadoopExternalSplit extends GridHadoopInputSplit {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private long off;
+
+ /**
+ * For {@link Externalizable}.
+ */
+ public HadoopExternalSplit() {
+ // No-op.
+ }
+
+ /**
+ * @param hosts Hosts.
+ * @param off Offset of this split in external file.
+ */
+ public HadoopExternalSplit(String[] hosts, long off) {
+ assert off >= 0 : off;
+ assert hosts != null;
+
+ this.hosts = hosts;
+ this.off = off;
+ }
+
+ /**
+ * @return Offset of this input split in external file.
+ */
+ public long offset() {
+ return off;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeLong(off);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ off = in.readLong();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ HadoopExternalSplit that = (HadoopExternalSplit) o;
+
+ return off == that.off;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return (int)(off ^ (off >>> 32));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopNativeCodeLoader.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopNativeCodeLoader.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopNativeCodeLoader.java
new file mode 100644
index 0000000..081182d
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopNativeCodeLoader.java
@@ -0,0 +1,74 @@
+/*
+ * 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.ignite.internal.processors.hadoop.v2;
+
+import org.apache.hadoop.classification.*;
+import org.apache.hadoop.conf.*;
+
+/**
+ * A fake helper to load the native hadoop code i.e. libhadoop.so.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class HadoopNativeCodeLoader {
+ /**
+ * Check if native-hadoop code is loaded for this platform.
+ *
+ * @return <code>true</code> if native-hadoop is loaded,
+ * else <code>false</code>
+ */
+ public static boolean isNativeCodeLoaded() {
+ return false;
+ }
+
+ /**
+ * Returns true only if this build was compiled with support for snappy.
+ */
+ public static boolean buildSupportsSnappy() {
+ return false;
+ }
+
+ /**
+ * @return Library name.
+ */
+ public static String getLibraryName() {
+ throw new IllegalStateException();
+ }
+
+ /**
+ * Return if native hadoop libraries, if present, can be used for this job.
+ * @param conf configuration
+ *
+ * @return <code>true</code> if native hadoop libraries, if present, can be
+ * used for this job; <code>false</code> otherwise.
+ */
+ public boolean getLoadNativeLibraries(Configuration conf) {
+ return false;
+ }
+
+ /**
+ * Set if native hadoop libraries, if present, can be used for this job.
+ *
+ * @param conf configuration
+ * @param loadNativeLibraries can native hadoop libraries be loaded
+ */
+ public void setLoadNativeLibraries(Configuration conf, boolean loadNativeLibraries) {
+ // No-op.
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSerializationWrapper.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSerializationWrapper.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSerializationWrapper.java
new file mode 100644
index 0000000..3e4a4c4
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSerializationWrapper.java
@@ -0,0 +1,133 @@
+/*
+ * 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.ignite.internal.processors.hadoop.v2;
+
+import org.apache.hadoop.io.serializer.*;
+import org.apache.ignite.*;
+import org.apache.ignite.internal.processors.hadoop.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+
+/**
+ * The wrapper around external serializer.
+ */
+public class HadoopSerializationWrapper<T> implements GridHadoopSerialization {
+ /** External serializer - writer. */
+ private final Serializer<T> serializer;
+
+ /** External serializer - reader. */
+ private final Deserializer<T> deserializer;
+
+ /** Data output for current write operation. */
+ private OutputStream currOut;
+
+ /** Data input for current read operation. */
+ private InputStream currIn;
+
+ /** Wrapper around current output to provide OutputStream interface. */
+ private final OutputStream outStream = new OutputStream() {
+ /** {@inheritDoc} */
+ @Override public void write(int b) throws IOException {
+ currOut.write(b);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(byte[] b, int off, int len) throws IOException {
+ currOut.write(b, off, len);
+ }
+ };
+
+ /** Wrapper around current input to provide InputStream interface. */
+ private final InputStream inStream = new InputStream() {
+ /** {@inheritDoc} */
+ @Override public int read() throws IOException {
+ return currIn.read();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int read(byte[] b, int off, int len) throws IOException {
+ return currIn.read(b, off, len);
+ }
+ };
+
+ /**
+ * @param serialization External serializer to wrap.
+ * @param cls The class to serialize.
+ */
+ public HadoopSerializationWrapper(Serialization<T> serialization, Class<T> cls) throws IgniteCheckedException {
+ assert cls != null;
+
+ serializer = serialization.getSerializer(cls);
+ deserializer = serialization.getDeserializer(cls);
+
+ try {
+ serializer.open(outStream);
+ deserializer.open(inStream);
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(DataOutput out, Object obj) throws IgniteCheckedException {
+ assert out != null;
+ assert obj != null;
+
+ try {
+ currOut = (OutputStream)out;
+
+ serializer.serialize((T)obj);
+
+ currOut = null;
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object read(DataInput in, @Nullable Object obj) throws IgniteCheckedException {
+ assert in != null;
+
+ try {
+ currIn = (InputStream)in;
+
+ T res = deserializer.deserialize((T) obj);
+
+ currIn = null;
+
+ return res;
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() throws IgniteCheckedException {
+ try {
+ serializer.close();
+ deserializer.close();
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopShutdownHookManager.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopShutdownHookManager.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopShutdownHookManager.java
new file mode 100644
index 0000000..454f90a
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopShutdownHookManager.java
@@ -0,0 +1,96 @@
+/*
+ * 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.ignite.internal.processors.hadoop.v2;
+
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Fake manager for shutdown hooks.
+ */
+public class HadoopShutdownHookManager {
+ /** */
+ private static final HadoopShutdownHookManager MGR = new HadoopShutdownHookManager();
+
+ /**
+ * Return <code>ShutdownHookManager</code> singleton.
+ *
+ * @return <code>ShutdownHookManager</code> singleton.
+ */
+ public static HadoopShutdownHookManager get() {
+ return MGR;
+ }
+
+ /** */
+ private Set<Runnable> hooks = Collections.synchronizedSet(new HashSet<Runnable>());
+
+ /** */
+ private AtomicBoolean shutdownInProgress = new AtomicBoolean(false);
+
+ /**
+ * Singleton.
+ */
+ private HadoopShutdownHookManager() {
+ // No-op.
+ }
+
+ /**
+ * Adds a shutdownHook with a priority, the higher the priority
+ * the earlier will run. ShutdownHooks with same priority run
+ * in a non-deterministic order.
+ *
+ * @param shutdownHook shutdownHook <code>Runnable</code>
+ * @param priority priority of the shutdownHook.
+ */
+ public void addShutdownHook(Runnable shutdownHook, int priority) {
+ if (shutdownHook == null)
+ throw new IllegalArgumentException("shutdownHook cannot be NULL");
+
+ hooks.add(shutdownHook);
+ }
+
+ /**
+ * Removes a shutdownHook.
+ *
+ * @param shutdownHook shutdownHook to remove.
+ * @return TRUE if the shutdownHook was registered and removed,
+ * FALSE otherwise.
+ */
+ public boolean removeShutdownHook(Runnable shutdownHook) {
+ return hooks.remove(shutdownHook);
+ }
+
+ /**
+ * Indicates if a shutdownHook is registered or not.
+ *
+ * @param shutdownHook shutdownHook to check if registered.
+ * @return TRUE/FALSE depending if the shutdownHook is is registered.
+ */
+ public boolean hasShutdownHook(Runnable shutdownHook) {
+ return hooks.contains(shutdownHook);
+ }
+
+ /**
+ * Indicates if shutdown is in progress or not.
+ *
+ * @return TRUE if the shutdown is in progress, otherwise FALSE.
+ */
+ public boolean isShutdownInProgress() {
+ return shutdownInProgress.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/1c4b00d4/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSplitWrapper.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSplitWrapper.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSplitWrapper.java
new file mode 100644
index 0000000..c73a8b0
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/v2/HadoopSplitWrapper.java
@@ -0,0 +1,118 @@
+/*
+ * 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.ignite.internal.processors.hadoop.v2;
+
+import org.apache.ignite.internal.processors.hadoop.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+
+/**
+ * The wrapper for native hadoop input splits.
+ *
+ * Warning!! This class must not depend on any Hadoop classes directly or indirectly.
+ */
+public class HadoopSplitWrapper extends GridHadoopInputSplit {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Native hadoop input split. */
+ private byte[] bytes;
+
+ /** */
+ private String clsName;
+
+ /** Internal ID */
+ private int id;
+
+ /**
+ * Creates new split wrapper.
+ */
+ public HadoopSplitWrapper() {
+ // No-op.
+ }
+
+ /**
+ * Creates new split wrapper.
+ *
+ * @param id Split ID.
+ * @param clsName Class name.
+ * @param bytes Serialized class.
+ * @param hosts Hosts where split is located.
+ */
+ public HadoopSplitWrapper(int id, String clsName, byte[] bytes, String[] hosts) {
+ assert hosts != null;
+ assert clsName != null;
+ assert bytes != null;
+
+ this.hosts = hosts;
+ this.id = id;
+
+ this.clsName = clsName;
+ this.bytes = bytes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(id);
+
+ out.writeUTF(clsName);
+ U.writeByteArray(out, bytes);
+ }
+
+ /**
+ * @return Class name.
+ */
+ public String className() {
+ return clsName;
+ }
+
+ /**
+ * @return Class bytes.
+ */
+ public byte[] bytes() {
+ return bytes;
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("unchecked")
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ id = in.readInt();
+
+ clsName = in.readUTF();
+ bytes = U.readByteArray(in);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ HadoopSplitWrapper that = (HadoopSplitWrapper)o;
+
+ return id == that.id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return id;
+ }
+}