You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mb...@apache.org on 2017/04/29 01:00:40 UTC
[2/3] asterixdb git commit: Remove Unused / Historical Hyracks Modules
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/HadoopPartitionerTuplePartitionComputerFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/HadoopPartitionerTuplePartitionComputerFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/HadoopPartitionerTuplePartitionComputerFactory.java
deleted file mode 100644
index 127eed9..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/HadoopPartitionerTuplePartitionComputerFactory.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.hyracks.dataflow.hadoop.data;
-
-import java.io.DataInputStream;
-
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.Partitioner;
-
-import org.apache.hyracks.api.comm.IFrameTupleAccessor;
-import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
-import org.apache.hyracks.api.dataflow.value.ITuplePartitionComputer;
-import org.apache.hyracks.api.dataflow.value.ITuplePartitionComputerFactory;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.util.ByteBufferInputStream;
-
-public class HadoopPartitionerTuplePartitionComputerFactory<K extends Writable, V extends Writable> extends
- AbstractClassBasedDelegate<Partitioner<K, V>> implements ITuplePartitionComputerFactory {
- private static final long serialVersionUID = 1L;
- private final ISerializerDeserializer<K> keyIO;
- private final ISerializerDeserializer<V> valueIO;
-
- public HadoopPartitionerTuplePartitionComputerFactory(Class<? extends Partitioner<K, V>> klass,
- ISerializerDeserializer<K> keyIO, ISerializerDeserializer<V> valueIO) {
- super(klass);
- this.keyIO = keyIO;
- this.valueIO = valueIO;
- }
-
- @Override
- public ITuplePartitionComputer createPartitioner() {
- return new ITuplePartitionComputer() {
- private final ByteBufferInputStream bbis = new ByteBufferInputStream();
- private final DataInputStream dis = new DataInputStream(bbis);
-
- @Override
- public int partition(IFrameTupleAccessor accessor, int tIndex, int nParts) throws HyracksDataException {
- int keyStart = accessor.getTupleStartOffset(tIndex) + accessor.getFieldSlotsLength()
- + accessor.getFieldStartOffset(tIndex, 0);
- bbis.setByteBuffer(accessor.getBuffer(), keyStart);
- K key = keyIO.deserialize(dis);
- int valueStart = accessor.getTupleStartOffset(tIndex) + accessor.getFieldSlotsLength()
- + accessor.getFieldStartOffset(tIndex, 1);
- bbis.setByteBuffer(accessor.getBuffer(), valueStart);
- V value = valueIO.deserialize(dis);
- return instance.getPartition(key, value, nParts);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyBinaryComparatorFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyBinaryComparatorFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyBinaryComparatorFactory.java
deleted file mode 100644
index b74fdde..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyBinaryComparatorFactory.java
+++ /dev/null
@@ -1,46 +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.hyracks.dataflow.hadoop.data;
-
-import org.apache.hadoop.io.RawComparator;
-
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.dataflow.common.util.ReflectionUtils;
-
-public class KeyBinaryComparatorFactory<T> implements IBinaryComparatorFactory {
- private static final long serialVersionUID = 1L;
-
- private Class<? extends RawComparator<T>> cmpClass;
-
- public KeyBinaryComparatorFactory(Class<? extends RawComparator<T>> cmpClass) {
- this.cmpClass = cmpClass;
- }
-
- @Override
- public IBinaryComparator createBinaryComparator() {
- final RawComparator<T> instance = ReflectionUtils.createInstance(cmpClass);
- return new IBinaryComparator() {
- @Override
- public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
- return instance.compare(b1, s1, l1, b2, s2, l2);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyComparatorFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyComparatorFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyComparatorFactory.java
deleted file mode 100644
index 01024bd..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/KeyComparatorFactory.java
+++ /dev/null
@@ -1,45 +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.hyracks.dataflow.hadoop.data;
-
-import org.apache.hadoop.io.RawComparator;
-
-import org.apache.hyracks.api.dataflow.value.IComparator;
-import org.apache.hyracks.api.dataflow.value.IComparatorFactory;
-import org.apache.hyracks.dataflow.common.util.ReflectionUtils;
-
-public class KeyComparatorFactory<T> implements IComparatorFactory<T> {
- private static final long serialVersionUID = 1L;
- private Class<? extends RawComparator<T>> cmpClass;
-
- public KeyComparatorFactory(Class<? extends RawComparator<T>> cmpClass) {
- this.cmpClass = cmpClass;
- }
-
- @Override
- public IComparator<T> createComparator() {
- final RawComparator<T> instance = ReflectionUtils.createInstance(cmpClass);
- return new IComparator<T>() {
- @Override
- public int compare(T o1, T o2) {
- return instance.compare(o1, o2);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/RawComparingComparatorFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/RawComparingComparatorFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/RawComparingComparatorFactory.java
deleted file mode 100644
index d2f0ead..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/RawComparingComparatorFactory.java
+++ /dev/null
@@ -1,47 +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.hyracks.dataflow.hadoop.data;
-
-import org.apache.hadoop.io.RawComparator;
-import org.apache.hadoop.io.WritableComparable;
-
-import org.apache.hyracks.api.dataflow.value.IComparator;
-import org.apache.hyracks.api.dataflow.value.IComparatorFactory;
-import org.apache.hyracks.dataflow.common.util.ReflectionUtils;
-
-public class RawComparingComparatorFactory<T> implements IComparatorFactory<WritableComparable<T>> {
- private Class<? extends RawComparator> klass;
-
- public RawComparingComparatorFactory(Class<? extends RawComparator> klass) {
- this.klass = klass;
- }
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public IComparator<WritableComparable<T>> createComparator() {
- final RawComparator instance = ReflectionUtils.createInstance(klass);
- return new IComparator<WritableComparable<T>>() {
- @Override
- public int compare(WritableComparable<T> o1, WritableComparable<T> o2) {
- return instance.compare(o1, o2);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/WritableComparingBinaryComparatorFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/WritableComparingBinaryComparatorFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/WritableComparingBinaryComparatorFactory.java
deleted file mode 100644
index 01f8755..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/data/WritableComparingBinaryComparatorFactory.java
+++ /dev/null
@@ -1,46 +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.hyracks.dataflow.hadoop.data;
-
-import org.apache.hadoop.io.RawComparator;
-
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.dataflow.common.util.ReflectionUtils;
-
-public class WritableComparingBinaryComparatorFactory<T> implements IBinaryComparatorFactory {
- private static final long serialVersionUID = 1L;
-
- private Class<? extends RawComparator<T>> cmpClass;
-
- public WritableComparingBinaryComparatorFactory(Class<? extends RawComparator<T>> cmpClass) {
- this.cmpClass = cmpClass;
- }
-
- @Override
- public IBinaryComparator createBinaryComparator() {
- final RawComparator<T> instance = ReflectionUtils.createInstance(cmpClass);
- return new IBinaryComparator() {
- @Override
- public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
- return instance.compare(b1, s1, l1, b2, s2, l2);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopHelper.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopHelper.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopHelper.java
deleted file mode 100644
index b248bff..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopHelper.java
+++ /dev/null
@@ -1,276 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.io.RawComparator;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.hadoop.mapreduce.Partitioner;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hadoop.util.ReflectionUtils;
-
-import org.apache.hyracks.api.context.IHyracksCommonContext;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
-import org.apache.hyracks.api.dataflow.value.ITuplePartitionComputerFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
-import org.apache.hyracks.dataflow.hadoop.data.HadoopNewPartitionerTuplePartitionComputerFactory;
-import org.apache.hyracks.dataflow.hadoop.data.WritableComparingBinaryComparatorFactory;
-import org.apache.hyracks.dataflow.hadoop.util.DatatypeHelper;
-import org.apache.hyracks.hdfs.ContextFactory;
-
-public class HadoopHelper {
- public static final int KEY_FIELD_INDEX = 0;
- public static final int VALUE_FIELD_INDEX = 1;
- public static final int BLOCKID_FIELD_INDEX = 2;
- private static final int[] KEY_SORT_FIELDS = new int[] { 0 };
-
- private MarshalledWritable<Configuration> mConfig;
- private Configuration config;
- private Job job;
-
- public HadoopHelper(MarshalledWritable<Configuration> mConfig) throws HyracksDataException {
- this.mConfig = mConfig;
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
- config = mConfig.get();
- config.setClassLoader(getClass().getClassLoader());
- job = new Job(config);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- }
-
- public RecordDescriptor getMapOutputRecordDescriptor() throws HyracksDataException {
- try {
- return new RecordDescriptor(
- new ISerializerDeserializer[] {
- DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputKeyClass()),
- DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputValueClass()), IntegerSerializerDeserializer.INSTANCE });
-
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- }
-
- public RecordDescriptor getMapOutputRecordDescriptorWithoutExtraFields() throws HyracksDataException {
- try {
- return new RecordDescriptor(
- new ISerializerDeserializer[] {
- DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputKeyClass()),
- DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputValueClass()) });
-
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- }
-
- public TaskAttemptContext createTaskAttemptContext(TaskAttemptID taId) throws HyracksDataException {
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(config.getClassLoader());
- return new ContextFactory().createContext(config, taId);
- } catch (HyracksDataException e) {
- e.printStackTrace();
- throw new HyracksDataException(e);
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- }
-
- public JobContext createJobContext() {
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(config.getClassLoader());
- return new ContextFactory().createJobContext(config);
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- }
-
- public <K1, V1, K2, V2> Mapper<K1, V1, K2, V2> getMapper() throws HyracksDataException {
- try {
- return (Mapper<K1, V1, K2, V2>) HadoopTools.newInstance(job.getMapperClass());
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- } catch (InstantiationException e) {
- throw new HyracksDataException(e);
- } catch (IllegalAccessException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public <K2, V2, K3, V3> Reducer<K2, V2, K3, V3> getReducer() throws HyracksDataException {
- try {
- return (Reducer<K2, V2, K3, V3>) HadoopTools.newInstance(job.getReducerClass());
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- } catch (InstantiationException e) {
- throw new HyracksDataException(e);
- } catch (IllegalAccessException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public <K2, V2> Reducer<K2, V2, K2, V2> getCombiner() throws HyracksDataException {
- try {
- return (Reducer<K2, V2, K2, V2>) HadoopTools.newInstance(job.getCombinerClass());
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- } catch (InstantiationException e) {
- throw new HyracksDataException(e);
- } catch (IllegalAccessException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public <K, V> InputFormat<K, V> getInputFormat() throws HyracksDataException {
- try {
- return (InputFormat<K, V>) ReflectionUtils.newInstance(job.getInputFormatClass(), config);
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public <K, V> List<InputSplit> getInputSplits() throws HyracksDataException {
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
- InputFormat<K, V> fmt = getInputFormat();
- JobContext jCtx = new ContextFactory().createJobContext(config);
- try {
- return fmt.getSplits(jCtx);
- } catch (IOException e) {
- throw new HyracksDataException(e);
- } catch (InterruptedException e) {
- throw new HyracksDataException(e);
- }
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- }
-
- public IBinaryComparatorFactory[] getSortComparatorFactories() {
- WritableComparingBinaryComparatorFactory comparatorFactory = new WritableComparingBinaryComparatorFactory(job
- .getSortComparator().getClass());
-
- return new IBinaryComparatorFactory[] { comparatorFactory };
- }
-
- public IBinaryComparatorFactory[] getGroupingComparatorFactories() {
- WritableComparingBinaryComparatorFactory comparatorFactory = new WritableComparingBinaryComparatorFactory(job
- .getGroupingComparator().getClass());
-
- return new IBinaryComparatorFactory[] { comparatorFactory };
- }
-
- public RawComparator<?> getRawGroupingComparator() {
- return job.getGroupingComparator();
- }
-
- public int getSortFrameLimit(IHyracksCommonContext ctx) {
- int sortMemory = job.getConfiguration().getInt("io.sort.mb", 100);
- return (int) (((long) sortMemory * 1024 * 1024) / ctx.getInitialFrameSize());
- }
-
- public Job getJob() {
- return job;
- }
-
- public MarshalledWritable<Configuration> getMarshalledConfiguration() {
- return mConfig;
- }
-
- public Configuration getConfiguration() {
- return config;
- }
-
- public ITuplePartitionComputerFactory getTuplePartitionComputer() throws HyracksDataException {
- int nReducers = job.getNumReduceTasks();
- try {
- return new HadoopNewPartitionerTuplePartitionComputerFactory<Writable, Writable>(
- (Class<? extends Partitioner<Writable, Writable>>) job.getPartitionerClass(),
- (ISerializerDeserializer<Writable>) DatatypeHelper
- .createSerializerDeserializer((Class<? extends Writable>) job.getMapOutputKeyClass()),
- (ISerializerDeserializer<Writable>) DatatypeHelper
- .createSerializerDeserializer((Class<? extends Writable>) job.getMapOutputValueClass()));
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public int[] getSortFields() {
- return KEY_SORT_FIELDS;
- }
-
- public <K> ISerializerDeserializer<K> getMapOutputKeySerializerDeserializer() {
- return (ISerializerDeserializer<K>) DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputKeyClass());
- }
-
- public <V> ISerializerDeserializer<V> getMapOutputValueSerializerDeserializer() {
- return (ISerializerDeserializer<V>) DatatypeHelper.createSerializerDeserializer((Class<? extends Writable>) job
- .getMapOutputValueClass());
- }
-
- public FileSystem getFilesystem() throws HyracksDataException {
- try {
- return FileSystem.get(config);
- } catch (IOException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public <K, V> OutputFormat<K, V> getOutputFormat() throws HyracksDataException {
- try {
- return (OutputFormat<K, V>) ReflectionUtils.newInstance(job.getOutputFormatClass(), config);
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public boolean hasCombiner() throws HyracksDataException {
- try {
- return job.getCombinerClass() != null;
- } catch (ClassNotFoundException e) {
- throw new HyracksDataException(e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopTools.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopTools.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopTools.java
deleted file mode 100644
index fb86385..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HadoopTools.java
+++ /dev/null
@@ -1,37 +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.hyracks.dataflow.hadoop.mapreduce;
-
-public class HadoopTools {
- public static Object newInstance(String className) throws ClassNotFoundException, InstantiationException,
- IllegalAccessException {
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(HadoopTools.class.getClassLoader());
- Class<?> clazz = Class.forName(className, true, HadoopTools.class.getClassLoader());
- return newInstance(clazz);
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- }
-
- public static Object newInstance(Class<?> clazz) throws InstantiationException, IllegalAccessException {
- return clazz.newInstance();
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HashPartitioningShuffleConnectorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HashPartitioningShuffleConnectorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HashPartitioningShuffleConnectorDescriptor.java
deleted file mode 100644
index 4892935..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/HashPartitioningShuffleConnectorDescriptor.java
+++ /dev/null
@@ -1,69 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.util.BitSet;
-
-import org.apache.hadoop.conf.Configuration;
-
-import org.apache.hyracks.api.comm.IFrameReader;
-import org.apache.hyracks.api.comm.IFrameWriter;
-import org.apache.hyracks.api.comm.IPartitionCollector;
-import org.apache.hyracks.api.comm.IPartitionWriterFactory;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.value.ITuplePartitionComputerFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.job.IConnectorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.base.AbstractMToNConnectorDescriptor;
-import org.apache.hyracks.dataflow.std.collectors.NonDeterministicChannelReader;
-import org.apache.hyracks.dataflow.std.collectors.PartitionCollector;
-import org.apache.hyracks.dataflow.std.connectors.PartitionDataWriter;
-
-public class HashPartitioningShuffleConnectorDescriptor extends AbstractMToNConnectorDescriptor {
- private static final long serialVersionUID = 1L;
-
- private final MarshalledWritable<Configuration> mConfig;
-
- public HashPartitioningShuffleConnectorDescriptor(IConnectorDescriptorRegistry spec, MarshalledWritable<Configuration> mConfig) {
- super(spec);
- this.mConfig = mConfig;
- }
-
- @Override
- public IFrameWriter createPartitioner(IHyracksTaskContext ctx, RecordDescriptor recordDesc,
- IPartitionWriterFactory edwFactory, int index, int nProducerPartitions, int nConsumerPartitions)
- throws HyracksDataException {
- HadoopHelper helper = new HadoopHelper(mConfig);
- ITuplePartitionComputerFactory tpcf = helper.getTuplePartitionComputer();
- return new PartitionDataWriter(ctx, nConsumerPartitions, edwFactory, recordDesc, tpcf.createPartitioner());
- }
-
- @Override
- public IPartitionCollector createPartitionCollector(IHyracksTaskContext ctx, RecordDescriptor recordDesc,
- int receiverIndex, int nProducerPartitions, int nConsumerPartitions) throws HyracksDataException {
- BitSet expectedPartitions = new BitSet();
- expectedPartitions.set(0, nProducerPartitions);
- NonDeterministicChannelReader channelReader = new NonDeterministicChannelReader(nProducerPartitions,
- expectedPartitions);
- IFrameReader frameReader = new ShuffleFrameReader(ctx, channelReader, mConfig);
- return new PartitionCollector(ctx, getConnectorId(), receiverIndex, expectedPartitions, frameReader,
- channelReader);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProvider.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProvider.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProvider.java
deleted file mode 100644
index 9cafaea..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProvider.java
+++ /dev/null
@@ -1,27 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import org.apache.hadoop.mapreduce.InputSplit;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-
-public interface IInputSplitProvider {
- public InputSplit next() throws HyracksDataException;
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProviderFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProviderFactory.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProviderFactory.java
deleted file mode 100644
index 87f7cd7..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/IInputSplitProviderFactory.java
+++ /dev/null
@@ -1,27 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.Serializable;
-
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-
-public interface IInputSplitProviderFactory extends Serializable {
- public IInputSplitProvider createInputSplitProvider(int id) throws HyracksDataException;
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/InputFileSplit.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/InputFileSplit.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/InputFileSplit.java
deleted file mode 100644
index d0e6ce2..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/InputFileSplit.java
+++ /dev/null
@@ -1,110 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-
-public class InputFileSplit extends InputSplit implements Writable {
- private Path file;
- private long start;
- private long length;
- private int blockId;
- private String[] hosts;
- private long scheduleTime;
-
- public InputFileSplit() {
- }
-
- public InputFileSplit(int blockId, Path file, long start, long length, String[] hosts, long schedule_time) {
- this.blockId = blockId;
- this.file = file;
- this.start = start;
- this.length = length;
- this.hosts = hosts;
- this.scheduleTime = schedule_time;
- }
-
- public int blockId() {
- return blockId;
- }
-
- public long scheduleTime() {
- return this.scheduleTime;
- }
-
- public Path getPath() {
- return file;
- }
-
- /** The position of the first byte in the file to process. */
- public long getStart() {
- return start;
- }
-
- /** The number of bytes in the file to process. */
- @Override
- public long getLength() {
- return length;
- }
-
- @Override
- public String toString() {
- return file + ":" + start + "+" + length;
- }
-
- @Override
- public void write(DataOutput out) throws IOException {
- Text.writeString(out, file.toString());
- out.writeLong(start);
- out.writeLong(length);
- out.writeInt(blockId);
- out.writeLong(this.scheduleTime);
- }
-
- @Override
- public void readFields(DataInput in) throws IOException {
- file = new Path(Text.readString(in));
- start = in.readLong();
- length = in.readLong();
- hosts = null;
- this.blockId = in.readInt();
- this.scheduleTime = in.readLong();
- }
-
- @Override
- public String[] getLocations() throws IOException {
- if (this.hosts == null) {
- return new String[] {};
- } else {
- return this.hosts;
- }
- }
-
- public FileSplit toFileSplit() {
- return new FileSplit(file, start, length, hosts);
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/KVIterator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/KVIterator.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/KVIterator.java
deleted file mode 100644
index 3f02279..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/KVIterator.java
+++ /dev/null
@@ -1,111 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.hadoop.io.DataInputBuffer;
-import org.apache.hadoop.mapred.RawKeyValueIterator;
-import org.apache.hadoop.util.Progress;
-
-import org.apache.hyracks.api.comm.IFrame;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-
-public class KVIterator implements RawKeyValueIterator {
- private final HadoopHelper helper;
- private FrameTupleAccessor accessor;
- private DataInputBuffer kBuffer;
- private DataInputBuffer vBuffer;
- private List<IFrame> buffers;
- private int bSize;
- private int bPtr;
- private int tIdx;
- private boolean eog;
-
- public KVIterator(HadoopHelper helper, RecordDescriptor recordDescriptor) {
- this.helper = helper;
- accessor = new FrameTupleAccessor(recordDescriptor);
- kBuffer = new DataInputBuffer();
- vBuffer = new DataInputBuffer();
- }
-
- void reset(List<IFrame> buffers, int bSize) {
- this.buffers = buffers;
- this.bSize = bSize;
- bPtr = 0;
- tIdx = 0;
- eog = false;
- if (bSize > 0) {
- accessor.reset(buffers.get(0).getBuffer());
- tIdx = -1;
- } else {
- eog = true;
- }
- }
-
- @Override
- public DataInputBuffer getKey() throws IOException {
- return kBuffer;
- }
-
- @Override
- public DataInputBuffer getValue() throws IOException {
- return vBuffer;
- }
-
- @Override
- public boolean next() throws IOException {
- while (true) {
- if (eog) {
- return false;
- }
- ++tIdx;
- if (accessor.getTupleCount() <= tIdx) {
- ++bPtr;
- if (bPtr >= bSize) {
- eog = true;
- continue;
- }
- tIdx = -1;
- accessor.reset(buffers.get(bPtr).getBuffer());
- continue;
- }
- kBuffer.reset(accessor.getBuffer().array(),
- accessor.getAbsoluteFieldStartOffset(tIdx, helper.KEY_FIELD_INDEX),
- accessor.getFieldLength(tIdx, helper.KEY_FIELD_INDEX));
- vBuffer.reset(accessor.getBuffer().array(),
- accessor.getAbsoluteFieldStartOffset(tIdx, helper.KEY_FIELD_INDEX),
- accessor.getFieldLength(tIdx, helper.VALUE_FIELD_INDEX));
- break;
- }
- return true;
- }
-
- @Override
- public void close() throws IOException {
-
- }
-
- @Override
- public Progress getProgress() {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MapperOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MapperOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MapperOperatorDescriptor.java
deleted file mode 100644
index 85cd34d..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MapperOperatorDescriptor.java
+++ /dev/null
@@ -1,271 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.DataOutput;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hyracks.api.comm.IFrame;
-import org.apache.hyracks.api.comm.IFrameWriter;
-import org.apache.hyracks.api.comm.VSizeFrame;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
-import org.apache.hyracks.dataflow.hadoop.util.MRContextUtil;
-import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
-import org.apache.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
-import org.apache.hyracks.dataflow.std.sort.Algorithm;
-import org.apache.hyracks.dataflow.std.sort.ExternalSortRunGenerator;
-import org.apache.hyracks.dataflow.std.sort.ExternalSortRunMerger;
-
-public class MapperOperatorDescriptor<K1 extends Writable, V1 extends Writable, K2 extends Writable, V2 extends Writable>
- extends AbstractSingleActivityOperatorDescriptor {
- private static final long serialVersionUID = 1L;
- private final int jobId;
- private final MarshalledWritable<Configuration> config;
- private final IInputSplitProviderFactory factory;
-
- public MapperOperatorDescriptor(IOperatorDescriptorRegistry spec, int jobId,
- MarshalledWritable<Configuration> config, IInputSplitProviderFactory factory) throws HyracksDataException {
- super(spec, 0, 1);
- this.jobId = jobId;
- this.config = config;
- this.factory = factory;
- HadoopHelper helper = new HadoopHelper(config);
- recordDescriptors[0] = helper.getMapOutputRecordDescriptor();
- }
-
- @SuppressWarnings("deprecation")
- @Override
- public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, final int partition, final int nPartitions)
- throws HyracksDataException {
- final HadoopHelper helper = new HadoopHelper(config);
- final Configuration conf = helper.getConfiguration();
- final Mapper<K1, V1, K2, V2> mapper = helper.getMapper();
- final InputFormat<K1, V1> inputFormat = helper.getInputFormat();
- final IInputSplitProvider isp = factory.createInputSplitProvider(partition);
- final TaskAttemptID taId = new TaskAttemptID("foo", jobId, true, partition, 0);
- final TaskAttemptContext taskAttemptContext = helper.createTaskAttemptContext(taId);
-
- final int framesLimit = helper.getSortFrameLimit(ctx);
- final IBinaryComparatorFactory[] comparatorFactories = helper.getSortComparatorFactories();
-
- class SortingRecordWriter extends RecordWriter<K2, V2> {
- private final ArrayTupleBuilder tb;
- private final IFrame frame;
- private final FrameTupleAppender fta;
- private ExternalSortRunGenerator runGen;
- private int blockId;
-
- public SortingRecordWriter() throws HyracksDataException {
- tb = new ArrayTupleBuilder(2);
- frame = new VSizeFrame(ctx);
- fta = new FrameTupleAppender(frame);
- }
-
- public void initBlock(int blockId) throws HyracksDataException {
- runGen = new ExternalSortRunGenerator(ctx, new int[] { 0 }, null, comparatorFactories,
- helper.getMapOutputRecordDescriptorWithoutExtraFields(), Algorithm.MERGE_SORT, framesLimit);
- this.blockId = blockId;
- }
-
- @Override
- public void close(TaskAttemptContext arg0) throws IOException, InterruptedException {
- }
-
- @Override
- public void write(K2 key, V2 value) throws IOException, InterruptedException {
- DataOutput dos = tb.getDataOutput();
- tb.reset();
- key.write(dos);
- tb.addFieldEndOffset();
- value.write(dos);
- tb.addFieldEndOffset();
- if (!fta.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- runGen.nextFrame(frame.getBuffer());
- fta.reset(frame, true);
- if (!fta.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- throw new HyracksDataException("Record size (" + tb.getSize() + ") larger than frame size ("
- + frame.getBuffer().capacity() + ")");
- }
- }
- }
-
- public void sortAndFlushBlock(final IFrameWriter writer) throws HyracksDataException {
- if (fta.getTupleCount() > 0) {
- runGen.nextFrame(frame.getBuffer());
- fta.reset(frame, true);
- }
- runGen.close();
- IFrameWriter delegatingWriter = new IFrameWriter() {
- private final FrameTupleAppender appender = new FrameTupleAppender(new VSizeFrame(ctx));
- private final FrameTupleAccessor fta = new FrameTupleAccessor(
- helper.getMapOutputRecordDescriptorWithoutExtraFields());
- private final ArrayTupleBuilder tb = new ArrayTupleBuilder(3);
-
- @Override
- public void open() throws HyracksDataException {
- }
-
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- fta.reset(buffer);
- int n = fta.getTupleCount();
- for (int i = 0; i < n; ++i) {
- tb.reset();
- tb.addField(fta, i, 0);
- tb.addField(fta, i, 1);
- try {
- tb.getDataOutput().writeInt(blockId);
- } catch (IOException e) {
- throw new HyracksDataException(e);
- }
- tb.addFieldEndOffset();
- if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- appender.write(writer, true);
- if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- throw new IllegalStateException();
- }
- }
- }
- }
-
- @Override
- public void close() throws HyracksDataException {
- appender.write(writer, true);
- }
-
- @Override
- public void fail() throws HyracksDataException {
- // TODO Auto-generated method stub
-
- }
- };
- if (helper.hasCombiner()) {
- Reducer<K2, V2, K2, V2> combiner = helper.getCombiner();
- TaskAttemptID ctaId = new TaskAttemptID("foo", jobId, true, partition, 0);
- TaskAttemptContext ctaskAttemptContext = helper.createTaskAttemptContext(taId);
- final IFrameWriter outputWriter = delegatingWriter;
- RecordWriter<K2, V2> recordWriter = new RecordWriter<K2, V2>() {
- private final FrameTupleAppender fta = new FrameTupleAppender(new VSizeFrame(ctx));
- private final ArrayTupleBuilder tb = new ArrayTupleBuilder(2);
-
- {
- outputWriter.open();
- }
-
- @Override
- public void write(K2 key, V2 value) throws IOException, InterruptedException {
- DataOutput dos = tb.getDataOutput();
- tb.reset();
- key.write(dos);
- tb.addFieldEndOffset();
- value.write(dos);
- tb.addFieldEndOffset();
- if (!fta.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- fta.write(outputWriter, true);
- if (!fta.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- throw new IllegalStateException();
- }
- }
- }
-
- @Override
- public void close(TaskAttemptContext context) throws IOException, InterruptedException {
- fta.write(outputWriter, true);
- }
- };
- delegatingWriter = new ReduceWriter<K2, V2, K2, V2>(ctx, helper,
- new int[] { HadoopHelper.KEY_FIELD_INDEX }, helper.getGroupingComparatorFactories(),
- helper.getMapOutputRecordDescriptorWithoutExtraFields(), combiner, recordWriter, ctaId,
- ctaskAttemptContext);
- }
- IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
- for (int i = 0; i < comparatorFactories.length; ++i) {
- comparators[i] = comparatorFactories[i].createBinaryComparator();
- }
- ExternalSortRunMerger merger = new ExternalSortRunMerger(ctx, runGen.getSorter(), runGen.getRuns(),
- new int[] { 0 }, comparators, null, helper.getMapOutputRecordDescriptorWithoutExtraFields(),
- framesLimit, delegatingWriter);
- merger.process();
- }
- }
-
- return new AbstractUnaryOutputSourceOperatorNodePushable() {
- @SuppressWarnings("unchecked")
- @Override
- public void initialize() throws HyracksDataException {
- try {
- writer.open();
- SortingRecordWriter recordWriter = new SortingRecordWriter();
- InputSplit split = null;
- int blockId = 0;
- while ((split = isp.next()) != null) {
- try {
- RecordReader<K1, V1> recordReader = inputFormat.createRecordReader(split,
- taskAttemptContext);
- ClassLoader ctxCL = Thread.currentThread().getContextClassLoader();
- try {
- Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
- recordReader.initialize(split, taskAttemptContext);
- } finally {
- Thread.currentThread().setContextClassLoader(ctxCL);
- }
- recordWriter.initBlock(blockId);
- Mapper<K1, V1, K2, V2>.Context mCtx = new MRContextUtil().createMapContext(conf, taId,
- recordReader, recordWriter, null, null, split);
- mapper.run(mCtx);
- recordReader.close();
- recordWriter.sortAndFlushBlock(writer);
- ++blockId;
- } catch (IOException e) {
- throw new HyracksDataException(e);
- } catch (InterruptedException e) {
- throw new HyracksDataException(e);
- }
- }
- } catch (Throwable th) {
- writer.fail();
- throw th;
- } finally {
- writer.close();
- }
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MarshalledWritable.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MarshalledWritable.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MarshalledWritable.java
deleted file mode 100644
index df53c39..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/MarshalledWritable.java
+++ /dev/null
@@ -1,55 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.Serializable;
-
-import org.apache.hadoop.io.Writable;
-
-public class MarshalledWritable<T extends Writable> implements Serializable {
- private static final long serialVersionUID = 1L;
-
- private byte[] bytes;
-
- public MarshalledWritable() {
- }
-
- public void set(T o) throws IOException {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(baos);
- dos.writeUTF(o.getClass().getName());
- o.write(dos);
- dos.close();
- bytes = baos.toByteArray();
- }
-
- public T get() throws Exception {
- ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
- DataInputStream dis = new DataInputStream(bais);
- String className = dis.readUTF();
- T o = (T) HadoopTools.newInstance(className);
- o.readFields(dis);
- return o;
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReduceWriter.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReduceWriter.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReduceWriter.java
deleted file mode 100644
index e64978f..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReduceWriter.java
+++ /dev/null
@@ -1,198 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.hadoop.io.RawComparator;
-import org.apache.hadoop.mapreduce.Counter;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hadoop.mapreduce.counters.GenericCounter;
-
-import org.apache.hyracks.api.comm.IFrame;
-import org.apache.hyracks.api.comm.IFrameWriter;
-import org.apache.hyracks.api.comm.VSizeFrame;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
-import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
-import org.apache.hyracks.dataflow.hadoop.util.MRContextUtil;
-
-public class ReduceWriter<K2, V2, K3, V3> implements IFrameWriter {
- private final IHyracksTaskContext ctx;
- private final HadoopHelper helper;
- private final int[] groupFields;
- private final FrameTupleAccessor accessor0;
- private final FrameTupleAccessor accessor1;
- private final IFrame copyFrame;
- private final IBinaryComparator[] comparators;
- private final KVIterator kvi;
- private final Reducer<K2, V2, K3, V3> reducer;
- private final RecordWriter<K3, V3> recordWriter;
- private final TaskAttemptID taId;
- private final TaskAttemptContext taskAttemptContext;
-
- private boolean first;
- private boolean groupStarted;
- private List<IFrame> group;
- private int bPtr;
- private FrameTupleAppender fta;
- private Counter keyCounter;
- private Counter valueCounter;
-
- public ReduceWriter(IHyracksTaskContext ctx, HadoopHelper helper, int[] groupFields,
- IBinaryComparatorFactory[] comparatorFactories, RecordDescriptor recordDescriptor,
- Reducer<K2, V2, K3, V3> reducer, RecordWriter<K3, V3> recordWriter, TaskAttemptID taId,
- TaskAttemptContext taskAttemptContext) throws HyracksDataException {
- this.ctx = ctx;
- this.helper = helper;
- this.groupFields = groupFields;
- accessor0 = new FrameTupleAccessor(recordDescriptor);
- accessor1 = new FrameTupleAccessor(recordDescriptor);
- copyFrame = new VSizeFrame(ctx);
- accessor1.reset(copyFrame.getBuffer());
- comparators = new IBinaryComparator[comparatorFactories.length];
- for (int i = 0; i < comparatorFactories.length; ++i) {
- comparators[i] = comparatorFactories[i].createBinaryComparator();
- }
- this.reducer = reducer;
- this.recordWriter = recordWriter;
- this.taId = taId;
- this.taskAttemptContext = taskAttemptContext;
-
- kvi = new KVIterator(helper, recordDescriptor);
- }
-
- @Override
- public void open() throws HyracksDataException {
- first = true;
- groupStarted = false;
- group = new ArrayList<>();
- bPtr = 0;
- group.add(new VSizeFrame(ctx));
- fta = new FrameTupleAppender();
- keyCounter = new GenericCounter();
- valueCounter = new GenericCounter();
- }
-
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- accessor0.reset(buffer);
- int nTuples = accessor0.getTupleCount();
- for (int i = 0; i < nTuples; ++i) {
- if (first) {
- groupInit();
- first = false;
- } else {
- if (i == 0) {
- accessor1.reset(copyFrame.getBuffer());
- switchGroupIfRequired(accessor1, accessor1.getTupleCount() - 1, accessor0, i);
- } else {
- switchGroupIfRequired(accessor0, i - 1, accessor0, i);
- }
- }
- accumulate(accessor0, i);
- }
- copyFrame.ensureFrameSize(buffer.capacity());
- FrameUtils.copyAndFlip(buffer, copyFrame.getBuffer());
- }
-
- private void accumulate(FrameTupleAccessor accessor, int tIndex) throws HyracksDataException {
- if (!fta.append(accessor, tIndex)) {
- ++bPtr;
- if (group.size() <= bPtr) {
- group.add(new VSizeFrame(ctx));
- }
- fta.reset(group.get(bPtr), true);
- if (!fta.append(accessor, tIndex)) {
- throw new HyracksDataException("Record size ("
- + (accessor.getTupleEndOffset(tIndex) - accessor.getTupleStartOffset(tIndex))
- + ") larger than frame size (" + group.get(bPtr).getBuffer().capacity() + ")");
- }
- }
- }
-
- private void switchGroupIfRequired(FrameTupleAccessor prevTupleAccessor, int prevTupleIndex,
- FrameTupleAccessor currTupleAccessor, int currTupleIndex) throws HyracksDataException {
- if (!sameGroup(prevTupleAccessor, prevTupleIndex, currTupleAccessor, currTupleIndex)) {
- reduce();
- groupInit();
- }
- }
-
- private void groupInit() throws HyracksDataException {
- groupStarted = true;
- bPtr = 0;
- fta.reset(group.get(0), true);
- }
-
- private void reduce() throws HyracksDataException {
- kvi.reset(group, bPtr + 1);
- try {
- Reducer<K2, V2, K3, V3>.Context rCtx = new MRContextUtil().createReduceContext(helper.getConfiguration(),
- taId, kvi, keyCounter, valueCounter, recordWriter, null, null, (RawComparator<K2>) helper
- .getRawGroupingComparator(), (Class<K2>) helper.getJob().getMapOutputKeyClass(),
- (Class<V2>) helper.getJob().getMapOutputValueClass());
- reducer.run(rCtx);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- groupStarted = false;
- }
-
- private boolean sameGroup(FrameTupleAccessor a1, int t1Idx, FrameTupleAccessor a2, int t2Idx)
- throws HyracksDataException {
- for (int i = 0; i < comparators.length; ++i) {
- int fIdx = groupFields[i];
- int s1 = a1.getTupleStartOffset(t1Idx) + a1.getFieldSlotsLength() + a1.getFieldStartOffset(t1Idx, fIdx);
- int l1 = a1.getFieldLength(t1Idx, fIdx);
- int s2 = a2.getTupleStartOffset(t2Idx) + a2.getFieldSlotsLength() + a2.getFieldStartOffset(t2Idx, fIdx);
- int l2 = a2.getFieldLength(t2Idx, fIdx);
- if (comparators[i].compare(a1.getBuffer().array(), s1, l1, a2.getBuffer().array(), s2, l2) != 0) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public void close() throws HyracksDataException {
- if (groupStarted) {
- reduce();
- }
- try {
- recordWriter.close(taskAttemptContext);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
- }
-
- @Override
- public void fail() throws HyracksDataException {
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReducerOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReducerOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReducerOperatorDescriptor.java
deleted file mode 100644
index d1bfbf7..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ReducerOperatorDescriptor.java
+++ /dev/null
@@ -1,96 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.nio.ByteBuffer;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-
-import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
-import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
-
-public class ReducerOperatorDescriptor<K2 extends Writable, V2 extends Writable, K3 extends Writable, V3 extends Writable>
- extends AbstractSingleActivityOperatorDescriptor {
- private static final long serialVersionUID = 1L;
-
- private final int jobId;
-
- private MarshalledWritable<Configuration> mConfig;
-
- public ReducerOperatorDescriptor(IOperatorDescriptorRegistry spec, int jobId, MarshalledWritable<Configuration> mConfig) {
- super(spec, 1, 0);
- this.jobId = jobId;
- this.mConfig = mConfig;
- }
-
- @Override
- public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
- IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
- final HadoopHelper helper = new HadoopHelper(mConfig);
- final Reducer<K2, V2, K3, V3> reducer = helper.getReducer();
- final RecordDescriptor recordDescriptor = helper.getMapOutputRecordDescriptor();
- final int[] groupFields = helper.getSortFields();
- IBinaryComparatorFactory[] groupingComparators = helper.getGroupingComparatorFactories();
-
- final TaskAttemptID taId = new TaskAttemptID("foo", jobId, false, partition, 0);
- final TaskAttemptContext taskAttemptContext = helper.createTaskAttemptContext(taId);
- final RecordWriter recordWriter;
- try {
- recordWriter = helper.getOutputFormat().getRecordWriter(taskAttemptContext);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
-
- final ReduceWriter<K2, V2, K3, V3> rw = new ReduceWriter<K2, V2, K3, V3>(ctx, helper, groupFields,
- groupingComparators, recordDescriptor, reducer, recordWriter, taId, taskAttemptContext);
-
- return new AbstractUnaryInputSinkOperatorNodePushable() {
- @Override
- public void open() throws HyracksDataException {
- rw.open();
- }
-
- @Override
- public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- rw.nextFrame(buffer);
- }
-
- @Override
- public void close() throws HyracksDataException {
- rw.close();
- }
-
- @Override
- public void fail() throws HyracksDataException {
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef360767/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ShuffleFrameReader.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ShuffleFrameReader.java b/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ShuffleFrameReader.java
deleted file mode 100644
index fb9c6fb..0000000
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-hadoop/src/main/java/org/apache/hyracks/dataflow/hadoop/mapreduce/ShuffleFrameReader.java
+++ /dev/null
@@ -1,219 +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.hyracks.dataflow.hadoop.mapreduce;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hyracks.api.comm.FrameHelper;
-import org.apache.hyracks.api.comm.IFrame;
-import org.apache.hyracks.api.comm.IFrameReader;
-import org.apache.hyracks.api.comm.NoShrinkVSizeFrame;
-import org.apache.hyracks.api.comm.VSizeFrame;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.data.std.primitive.IntegerPointable;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
-import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
-import org.apache.hyracks.dataflow.common.io.RunFileReader;
-import org.apache.hyracks.dataflow.common.io.RunFileWriter;
-import org.apache.hyracks.dataflow.std.collectors.NonDeterministicChannelReader;
-import org.apache.hyracks.dataflow.std.sort.ExternalSortRunMerger;
-import org.apache.hyracks.dataflow.std.structures.RunAndMaxFrameSizePair;
-
-public class ShuffleFrameReader implements IFrameReader {
- private final IHyracksTaskContext ctx;
- private final NonDeterministicChannelReader channelReader;
- private final HadoopHelper helper;
- private final RecordDescriptor recordDescriptor;
- private final IFrame vframe;
- private List<RunFileWriter> runFileWriters;
- private List<Integer> runFileMaxFrameSize;
- private RunFileReader reader;
-
- public ShuffleFrameReader(IHyracksTaskContext ctx, NonDeterministicChannelReader channelReader,
- MarshalledWritable<Configuration> mConfig) throws HyracksDataException {
- this.ctx = ctx;
- this.channelReader = channelReader;
- this.helper = new HadoopHelper(mConfig);
- this.recordDescriptor = helper.getMapOutputRecordDescriptor();
- this.vframe = new NoShrinkVSizeFrame(ctx);
- }
-
- @Override
- public void open() throws HyracksDataException {
- channelReader.open();
- int nSenders = channelReader.getSenderPartitionCount();
- runFileWriters = new ArrayList<RunFileWriter>();
- runFileMaxFrameSize = new ArrayList<>();
- RunInfo[] infos = new RunInfo[nSenders];
- FrameTupleAccessor accessor = new FrameTupleAccessor(recordDescriptor);
- while (true) {
- int entry = channelReader.findNextSender();
- if (entry < 0) {
- break;
- }
- RunInfo info = infos[entry];
- ByteBuffer netBuffer = channelReader.getNextBuffer(entry);
- netBuffer.clear();
- int nBlocks = FrameHelper.deserializeNumOfMinFrame(netBuffer);
-
- if (nBlocks > 1) {
- netBuffer = getCompleteBuffer(nBlocks, netBuffer, entry);
- }
-
- accessor.reset(netBuffer, 0, netBuffer.limit());
- int nTuples = accessor.getTupleCount();
- for (int i = 0; i < nTuples; ++i) {
- int tBlockId = IntegerPointable.getInteger(accessor.getBuffer().array(),
- accessor.getAbsoluteFieldStartOffset(i, HadoopHelper.BLOCKID_FIELD_INDEX));
- if (info == null) {
- info = new RunInfo();
- info.reset(tBlockId);
- infos[entry] = info;
- } else if (info.blockId != tBlockId) {
- info.close();
- info.reset(tBlockId);
- }
- info.write(accessor, i);
- }
-
- if (nBlocks == 1) {
- channelReader.recycleBuffer(entry, netBuffer);
- }
- }
- for (int i = 0; i < infos.length; ++i) {
- RunInfo info = infos[i];
- if (info != null) {
- info.close();
- }
- }
-
- FileReference outFile = ctx.createManagedWorkspaceFile(ShuffleFrameReader.class.getName() + ".run");
- int framesLimit = helper.getSortFrameLimit(ctx);
- IBinaryComparatorFactory[] comparatorFactories = helper.getSortComparatorFactories();
- IBinaryComparator[] comparators = new IBinaryComparator[comparatorFactories.length];
- for (int i = 0; i < comparatorFactories.length; ++i) {
- comparators[i] = comparatorFactories[i].createBinaryComparator();
- }
- List<RunAndMaxFrameSizePair> runs = new LinkedList<>();
- for (int i = 0; i < runFileWriters.size(); i++) {
- runs.add(new RunAndMaxFrameSizePair(runFileWriters.get(i).createDeleteOnCloseReader(), runFileMaxFrameSize
- .get(i)));
- }
- RunFileWriter rfw = new RunFileWriter(outFile, ctx.getIOManager());
- ExternalSortRunMerger merger = new ExternalSortRunMerger(ctx, null, runs, new int[] { 0 }, comparators, null,
- recordDescriptor, framesLimit, rfw);
- merger.process();
-
- reader = rfw.createDeleteOnCloseReader();
- reader.open();
- }
-
- private ByteBuffer getCompleteBuffer(int nBlocks, ByteBuffer netBuffer, int entry) throws HyracksDataException {
- vframe.reset();
- vframe.ensureFrameSize(vframe.getMinSize() * nBlocks);
- FrameUtils.copyWholeFrame(netBuffer, vframe.getBuffer());
- channelReader.recycleBuffer(entry, netBuffer);
- for (int i = 1; i < nBlocks; ++i) {
- netBuffer = channelReader.getNextBuffer(entry);
- netBuffer.clear();
- vframe.getBuffer().put(netBuffer);
- channelReader.recycleBuffer(entry, netBuffer);
- }
- if (vframe.getBuffer().hasRemaining()) { // bigger frame
- FrameHelper.clearRemainingFrame(vframe.getBuffer(), vframe.getBuffer().position());
- }
- vframe.getBuffer().flip();
- return vframe.getBuffer();
- }
-
- @Override
- public boolean nextFrame(IFrame frame) throws HyracksDataException {
- return reader.nextFrame(frame);
- }
-
- @Override
- public void close() throws HyracksDataException {
- reader.close();
- }
-
- private class RunInfo {
- private final IFrame buffer;
- private final FrameTupleAppender fta;
-
- private FileReference file;
- private RunFileWriter rfw;
- private int blockId;
- private int maxFrameSize = ctx.getInitialFrameSize();
-
- public RunInfo() throws HyracksDataException {
- buffer = new VSizeFrame(ctx);
- fta = new FrameTupleAppender();
- }
-
- public void reset(int blockId) throws HyracksDataException {
- this.blockId = blockId;
- this.maxFrameSize = ctx.getInitialFrameSize();
- fta.reset(buffer, true);
- try {
- file = ctx.createManagedWorkspaceFile(ShuffleFrameReader.class.getName() + ".run");
- rfw = new RunFileWriter(file, ctx.getIOManager());
- } catch (IOException e) {
- throw new HyracksDataException(e);
- }
- }
-
- public void write(FrameTupleAccessor accessor, int tIdx) throws HyracksDataException {
- if (!fta.append(accessor, tIdx)) {
- flush();
- if (!fta.append(accessor, tIdx)) {
- throw new HyracksDataException("Record size ("
- + (accessor.getTupleEndOffset(tIdx) - accessor.getTupleStartOffset(tIdx))
- + ") larger than frame size (" + fta.getBuffer().capacity() + ")");
- }
- }
- }
-
- public void close() throws HyracksDataException {
- flush();
- rfw.close();
- runFileWriters.add(rfw);
- runFileMaxFrameSize.add(maxFrameSize);
- }
-
- private void flush() throws HyracksDataException {
- if (fta.getTupleCount() <= 0) {
- return;
- }
- maxFrameSize = buffer.getFrameSize() > maxFrameSize ? buffer.getFrameSize() : maxFrameSize;
- rfw.nextFrame((ByteBuffer) buffer.getBuffer().clear());
- fta.reset(buffer, true);
- }
- }
-}