You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@tez.apache.org by "okumin (via GitHub)" <gi...@apache.org> on 2023/12/11 13:16:50 UTC
[PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
okumin opened a new pull request, #320:
URL: https://github.com/apache/tez/pull/320
This is a sample application that uses FairShuffleVertexManager.
We can use it like this.
```
$ hadoop jar /opt/tez/tez-examples-*.jar demuxerdatagen hdfs:///tmp/in 30 200
$ hadoop jar /opt/tez/tez-examples-*.jar demuxer hdfs:///tmp/in hdfs:///tmp/out 100
$ hadoop jar /opt/tez/tez-examples-*.jar demuxer hdfs:///tmp/in hdfs:///tmp/out_precise 100 true
```
The following are the results on my machine. We can see skewed keys are written by multiple reducers.
```
zookage@client-node-0:~$ hdfs dfs -ls /tmp/out
Found 33 items
-rw-r--r-- 3 hdfs supergroup 0 2023-12-11 09:20 /tmp/out/_SUCCESS
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00000-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00001-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00002-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00003-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00004-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00005-r-00007
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00006-r-00008
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00007-r-00008
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:18 /tmp/out/category-00008-r-00008
-rw-r--r-- 3 hdfs supergroup 6000 2023-12-11 09:18 /tmp/out/category-00009-r-00008
-rw-r--r-- 3 hdfs supergroup 15000 2023-12-11 09:17 /tmp/out/category-00010-r-00000
-rw-r--r-- 3 hdfs supergroup 30000 2023-12-11 09:17 /tmp/out/category-00011-r-00000
-rw-r--r-- 3 hdfs supergroup 60000 2023-12-11 09:17 /tmp/out/category-00012-r-00000
-rw-r--r-- 3 hdfs supergroup 120000 2023-12-11 09:17 /tmp/out/category-00013-r-00000
-rw-r--r-- 3 hdfs supergroup 243000 2023-12-11 09:17 /tmp/out/category-00014-r-00000
-rw-r--r-- 3 hdfs supergroup 489000 2023-12-11 09:17 /tmp/out/category-00015-r-00000
-rw-r--r-- 3 hdfs supergroup 981000 2023-12-11 09:17 /tmp/out/category-00016-r-00001
-rw-r--r-- 3 hdfs supergroup 1965000 2023-12-11 09:17 /tmp/out/category-00017-r-00001
-rw-r--r-- 3 hdfs supergroup 3930000 2023-12-11 09:17 /tmp/out/category-00018-r-00001
-rw-r--r-- 3 hdfs supergroup 7863000 2023-12-11 09:17 /tmp/out/category-00019-r-00001
-rw-r--r-- 3 hdfs supergroup 15726000 2023-12-11 09:17 /tmp/out/category-00020-r-00001
-rw-r--r-- 3 hdfs supergroup 31455000 2023-12-11 09:17 /tmp/out/category-00021-r-00001
-rw-r--r-- 3 hdfs supergroup 62913000 2023-12-11 09:20 /tmp/out/category-00022-r-00002
-rw-r--r-- 3 hdfs supergroup 125829000 2023-12-11 09:20 /tmp/out/category-00023-r-00002
-rw-r--r-- 3 hdfs supergroup 251658000 2023-12-11 09:20 /tmp/out/category-00024-r-00002
-rw-r--r-- 3 hdfs supergroup 503316000 2023-12-11 09:20 /tmp/out/category-00025-r-00002
-rw-r--r-- 3 hdfs supergroup 1006632000 2023-12-11 09:20 /tmp/out/category-00026-r-00002
-rw-r--r-- 3 hdfs supergroup 2013264000 2023-12-11 09:20 /tmp/out/category-00027-r-00002
-rw-r--r-- 3 hdfs supergroup 2093796120 2023-12-11 09:18 /tmp/out/category-00028-r-00003
-rw-r--r-- 3 hdfs supergroup 1932734880 2023-12-11 09:18 /tmp/out/category-00028-r-00004
-rw-r--r-- 3 hdfs supergroup 4187592240 2023-12-11 09:20 /tmp/out/category-00029-r-00005
-rw-r--r-- 3 hdfs supergroup 3865469760 2023-12-11 09:20 /tmp/out/category-00029-r-00006
```
```
zookage@client-node-0:~$ hdfs dfs -ls /tmp/out_precise
Found 39 items
-rw-r--r-- 3 hdfs supergroup 0 2023-12-11 09:26 /tmp/out_precise/_SUCCESS
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00000-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00001-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00002-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00003-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00004-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00005-r-00016
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00006-r-00017
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00007-r-00017
-rw-r--r-- 3 hdfs supergroup 3000 2023-12-11 09:24 /tmp/out_precise/category-00008-r-00017
-rw-r--r-- 3 hdfs supergroup 6000 2023-12-11 09:24 /tmp/out_precise/category-00009-r-00017
-rw-r--r-- 3 hdfs supergroup 15000 2023-12-11 09:23 /tmp/out_precise/category-00010-r-00000
-rw-r--r-- 3 hdfs supergroup 30000 2023-12-11 09:23 /tmp/out_precise/category-00011-r-00000
-rw-r--r-- 3 hdfs supergroup 60000 2023-12-11 09:23 /tmp/out_precise/category-00012-r-00000
-rw-r--r-- 3 hdfs supergroup 120000 2023-12-11 09:23 /tmp/out_precise/category-00013-r-00000
-rw-r--r-- 3 hdfs supergroup 243000 2023-12-11 09:23 /tmp/out_precise/category-00014-r-00000
-rw-r--r-- 3 hdfs supergroup 489000 2023-12-11 09:23 /tmp/out_precise/category-00015-r-00000
-rw-r--r-- 3 hdfs supergroup 981000 2023-12-11 09:24 /tmp/out_precise/category-00016-r-00001
-rw-r--r-- 3 hdfs supergroup 1965000 2023-12-11 09:24 /tmp/out_precise/category-00017-r-00001
-rw-r--r-- 3 hdfs supergroup 3930000 2023-12-11 09:24 /tmp/out_precise/category-00018-r-00001
-rw-r--r-- 3 hdfs supergroup 7863000 2023-12-11 09:24 /tmp/out_precise/category-00019-r-00001
-rw-r--r-- 3 hdfs supergroup 15726000 2023-12-11 09:24 /tmp/out_precise/category-00020-r-00001
-rw-r--r-- 3 hdfs supergroup 31455000 2023-12-11 09:24 /tmp/out_precise/category-00021-r-00001
-rw-r--r-- 3 hdfs supergroup 62913000 2023-12-11 09:24 /tmp/out_precise/category-00022-r-00002
-rw-r--r-- 3 hdfs supergroup 125829000 2023-12-11 09:24 /tmp/out_precise/category-00023-r-00002
-rw-r--r-- 3 hdfs supergroup 251658000 2023-12-11 09:24 /tmp/out_precise/category-00024-r-00002
-rw-r--r-- 3 hdfs supergroup 503316000 2023-12-11 09:24 /tmp/out_precise/category-00025-r-00003
-rw-r--r-- 3 hdfs supergroup 1006632000 2023-12-11 09:24 /tmp/out_precise/category-00026-r-00004
-rw-r--r-- 3 hdfs supergroup 1046897280 2023-12-11 09:24 /tmp/out_precise/category-00027-r-00005
-rw-r--r-- 3 hdfs supergroup 966366720 2023-12-11 09:24 /tmp/out_precise/category-00027-r-00006
-rw-r--r-- 3 hdfs supergroup 1308622575 2023-12-11 09:24 /tmp/out_precise/category-00028-r-00007
-rw-r--r-- 3 hdfs supergroup 1308622575 2023-12-11 09:24 /tmp/out_precise/category-00028-r-00008
-rw-r--r-- 3 hdfs supergroup 1409285850 2023-12-11 09:24 /tmp/out_precise/category-00028-r-00009
-rw-r--r-- 3 hdfs supergroup 1046898060 2023-12-11 09:24 /tmp/out_precise/category-00029-r-00010
-rw-r--r-- 3 hdfs supergroup 1046898060 2023-12-11 09:24 /tmp/out_precise/category-00029-r-00011
-rw-r--r-- 3 hdfs supergroup 1570347090 2023-12-11 09:26 /tmp/out_precise/category-00029-r-00012
-rw-r--r-- 3 hdfs supergroup 1570347090 2023-12-11 09:26 /tmp/out_precise/category-00029-r-00013
-rw-r--r-- 3 hdfs supergroup 1570347090 2023-12-11 09:26 /tmp/out_precise/category-00029-r-00014
-rw-r--r-- 3 hdfs supergroup 1248224610 2023-12-11 09:26 /tmp/out_precise/category-00029-r-00015
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #320:
URL: https://github.com/apache/tez/pull/320#discussion_r1703083974
##########
tez-examples/src/main/java/org/apache/tez/examples/Demuxer.java:
##########
@@ -0,0 +1,231 @@
+/**
+ * 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.tez.examples;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.GroupInputEdge;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager.FairRoutingType;
+import org.apache.tez.examples.HashJoinExample.ForwardingProcessor;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MultiMROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueWriterWithBasePath;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration.ReportPartitionStats;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValuesInput;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple example of a demuxer tolerant of data skew problems. The example scans the source files,
+ * partitions records by values, and write them in /path/to/{value}-r-{task id}.
+ */
+public class Demuxer extends TezExampleBase {
+ private static final Logger LOG = LoggerFactory.getLogger(Demuxer.class);
+
+ private static final String DEMUXER_OUTPUT = "demuxerOutput";
+
+ public static void main(String[] args) throws Exception {
+ Demuxer job = new Demuxer();
+ int status = ToolRunner.run(new Configuration(), job, args);
+ System.exit(status);
+ }
+
+ @Override
+ protected void printUsage() {
+ System.err.println("Usage: "
+ + "demuxer <inputPath> <outPath> <numPartitions> [isPrecise(default false)]");
+ }
+
+ @Override
+ protected int validateArgs(String[] otherArgs) {
+ if (otherArgs.length < 3 || otherArgs.length > 4) {
+ return 2;
+ }
+ return 0;
+ }
+
+ @Override
+ protected int runJob(String[] args, TezConfiguration tezConf, TezClient tezClient)
+ throws Exception {
+ String inputDirs = args[0];
+ String outputDir = args[1];
+ int numPartitions = Integer.parseInt(args[2]);
+ if (args.length == 4 && Boolean.parseBoolean(args[3])) {
+ tezConf.set(
+ TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS,
+ ReportPartitionStats.PRECISE.getType());
+ }
+
+ List<Path> inputPaths = Arrays
+ .stream(inputDirs.split(","))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ Path outputPath = new Path(outputDir);
+
+ FileSystem fs = outputPath.getFileSystem(tezConf);
+ outputPath = fs.makeQualified(outputPath);
+ if (fs.exists(outputPath)) {
+ System.err.println("Output directory: " + outputDir + " already exists");
+ return 3;
+ }
+ DAG dag = inputPaths.size() == 1
+ ? createDag(tezConf, inputPaths.get(0), outputPath, numPartitions)
+ : createDagWithUnion(tezConf, inputPaths, outputPath, numPartitions);
+ LOG.info("Running Demuxer");
+ return runDag(dag, isCountersLog(), LOG);
+ }
+
+ private DAG createDag(TezConfiguration tezConf, Path inputPath, Path outputPath,
+ int numPartitions) {
+ Vertex inputVertex = createInputVertex(tezConf, "input", inputPath);
+ Vertex demuxVertex = createDemuxVertex(tezConf, outputPath, numPartitions);
+ OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig
+ .newBuilder(
+ Text.class.getName(),
+ NullWritable.class.getName(),
+ HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build();
+ return DAG
+ .create("Demuxer")
+ .addVertex(inputVertex)
+ .addVertex(demuxVertex)
+ .addEdge(Edge.create(inputVertex, demuxVertex, edgeConf.createDefaultEdgeProperty()));
+ }
+
+ private DAG createDagWithUnion(TezConfiguration tezConf, List<Path> inputPaths, Path outputPath,
Review Comment:
https://github.com/apache/tez/pull/306 was closed due to some reasons
is this PR suitable to be adapted to work without that patch? if so, still interested in seeing this in the codebase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #320:
URL: https://github.com/apache/tez/pull/320#discussion_r1422456872
##########
tez-examples/src/main/java/org/apache/tez/examples/Demuxer.java:
##########
@@ -0,0 +1,231 @@
+/**
+ * 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.tez.examples;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.GroupInputEdge;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager.FairRoutingType;
+import org.apache.tez.examples.HashJoinExample.ForwardingProcessor;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MultiMROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueWriterWithBasePath;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration.ReportPartitionStats;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValuesInput;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple example of a demuxer tolerant of data skew problems. The example scans the source files,
+ * partitions records by values, and write them in /path/to/{value}-r-{task id}.
+ */
+public class Demuxer extends TezExampleBase {
+ private static final Logger LOG = LoggerFactory.getLogger(Demuxer.class);
+
+ private static final String DEMUXER_OUTPUT = "demuxerOutput";
+
+ public static void main(String[] args) throws Exception {
+ Demuxer job = new Demuxer();
+ int status = ToolRunner.run(new Configuration(), job, args);
+ System.exit(status);
+ }
+
+ @Override
+ protected void printUsage() {
+ System.err.println("Usage: "
+ + "demuxer <inputPath> <outPath> <numPartitions> [isPrecise(default false)]");
+ }
+
+ @Override
+ protected int validateArgs(String[] otherArgs) {
+ if (otherArgs.length < 3 || otherArgs.length > 4) {
+ return 2;
+ }
+ return 0;
+ }
+
+ @Override
+ protected int runJob(String[] args, TezConfiguration tezConf, TezClient tezClient)
+ throws Exception {
+ String inputDirs = args[0];
+ String outputDir = args[1];
+ int numPartitions = Integer.parseInt(args[2]);
+ if (args.length == 4 && Boolean.parseBoolean(args[3])) {
+ tezConf.set(
+ TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS,
+ ReportPartitionStats.PRECISE.getType());
+ }
+
+ List<Path> inputPaths = Arrays
+ .stream(inputDirs.split(","))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ Path outputPath = new Path(outputDir);
+
+ FileSystem fs = outputPath.getFileSystem(tezConf);
+ outputPath = fs.makeQualified(outputPath);
+ if (fs.exists(outputPath)) {
+ System.err.println("Output directory: " + outputDir + " already exists");
+ return 3;
+ }
+ DAG dag = inputPaths.size() == 1
+ ? createDag(tezConf, inputPaths.get(0), outputPath, numPartitions)
+ : createDagWithUnion(tezConf, inputPaths, outputPath, numPartitions);
+ LOG.info("Running Demuxer");
+ return runDag(dag, isCountersLog(), LOG);
+ }
+
+ private DAG createDag(TezConfiguration tezConf, Path inputPath, Path outputPath,
+ int numPartitions) {
+ Vertex inputVertex = createInputVertex(tezConf, "input", inputPath);
+ Vertex demuxVertex = createDemuxVertex(tezConf, outputPath, numPartitions);
+ OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig
+ .newBuilder(
+ Text.class.getName(),
+ NullWritable.class.getName(),
+ HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build();
+ return DAG
+ .create("Demuxer")
+ .addVertex(inputVertex)
+ .addVertex(demuxVertex)
+ .addEdge(Edge.create(inputVertex, demuxVertex, edgeConf.createDefaultEdgeProperty()));
+ }
+
+ private DAG createDagWithUnion(TezConfiguration tezConf, List<Path> inputPaths, Path outputPath,
Review Comment:
Note: This case doesn't work yet until we merge https://github.com/apache/tez/pull/306
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "okumin (via GitHub)" <gi...@apache.org>.
okumin commented on code in PR #320:
URL: https://github.com/apache/tez/pull/320#discussion_r1703177496
##########
tez-examples/src/main/java/org/apache/tez/examples/Demuxer.java:
##########
@@ -0,0 +1,231 @@
+/**
+ * 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.tez.examples;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.GroupInputEdge;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager.FairRoutingType;
+import org.apache.tez.examples.HashJoinExample.ForwardingProcessor;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MultiMROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueWriterWithBasePath;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration.ReportPartitionStats;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValuesInput;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple example of a demuxer tolerant of data skew problems. The example scans the source files,
+ * partitions records by values, and write them in /path/to/{value}-r-{task id}.
+ */
+public class Demuxer extends TezExampleBase {
+ private static final Logger LOG = LoggerFactory.getLogger(Demuxer.class);
+
+ private static final String DEMUXER_OUTPUT = "demuxerOutput";
+
+ public static void main(String[] args) throws Exception {
+ Demuxer job = new Demuxer();
+ int status = ToolRunner.run(new Configuration(), job, args);
+ System.exit(status);
+ }
+
+ @Override
+ protected void printUsage() {
+ System.err.println("Usage: "
+ + "demuxer <inputPath> <outPath> <numPartitions> [isPrecise(default false)]");
+ }
+
+ @Override
+ protected int validateArgs(String[] otherArgs) {
+ if (otherArgs.length < 3 || otherArgs.length > 4) {
+ return 2;
+ }
+ return 0;
+ }
+
+ @Override
+ protected int runJob(String[] args, TezConfiguration tezConf, TezClient tezClient)
+ throws Exception {
+ String inputDirs = args[0];
+ String outputDir = args[1];
+ int numPartitions = Integer.parseInt(args[2]);
+ if (args.length == 4 && Boolean.parseBoolean(args[3])) {
+ tezConf.set(
+ TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS,
+ ReportPartitionStats.PRECISE.getType());
+ }
+
+ List<Path> inputPaths = Arrays
+ .stream(inputDirs.split(","))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ Path outputPath = new Path(outputDir);
+
+ FileSystem fs = outputPath.getFileSystem(tezConf);
+ outputPath = fs.makeQualified(outputPath);
+ if (fs.exists(outputPath)) {
+ System.err.println("Output directory: " + outputDir + " already exists");
+ return 3;
+ }
+ DAG dag = inputPaths.size() == 1
+ ? createDag(tezConf, inputPaths.get(0), outputPath, numPartitions)
+ : createDagWithUnion(tezConf, inputPaths, outputPath, numPartitions);
+ LOG.info("Running Demuxer");
+ return runDag(dag, isCountersLog(), LOG);
+ }
+
+ private DAG createDag(TezConfiguration tezConf, Path inputPath, Path outputPath,
+ int numPartitions) {
+ Vertex inputVertex = createInputVertex(tezConf, "input", inputPath);
+ Vertex demuxVertex = createDemuxVertex(tezConf, outputPath, numPartitions);
+ OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig
+ .newBuilder(
+ Text.class.getName(),
+ NullWritable.class.getName(),
+ HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build();
+ return DAG
+ .create("Demuxer")
+ .addVertex(inputVertex)
+ .addVertex(demuxVertex)
+ .addEdge(Edge.create(inputVertex, demuxVertex, edgeConf.createDefaultEdgeProperty()));
+ }
+
+ private DAG createDagWithUnion(TezConfiguration tezConf, List<Path> inputPaths, Path outputPath,
Review Comment:
@abstractdog Thanks for asking. I closed #306 and still need to implement an alternative solution. At this point, I removed the case using UNION from the sample.
https://github.com/apache/tez/pull/320/commits/9e0966541383e305d4505f23178aaebfc90181b5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #320:
URL: https://github.com/apache/tez/pull/320#issuecomment-1850122576
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 13m 36s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| -1 :x: | test4tests | 0m 0s | The patch doesn't appear to include any new or modified tests. Please justify why no new tests are needed for this patch. Also please list what manual steps were performed to verify this patch. |
||| _ master Compile Tests _ |
| +1 :green_heart: | mvninstall | 16m 0s | master passed |
| +1 :green_heart: | compile | 0m 18s | master passed with JDK Ubuntu-11.0.21+9-post-Ubuntu-0ubuntu122.04 |
| +1 :green_heart: | compile | 0m 15s | master passed with JDK Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| +1 :green_heart: | checkstyle | 1m 26s | master passed |
| +1 :green_heart: | javadoc | 0m 28s | master passed with JDK Ubuntu-11.0.21+9-post-Ubuntu-0ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 14s | master passed with JDK Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| +0 :ok: | spotbugs | 1m 4s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 1m 2s | master passed |
||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 11s | the patch passed |
| +1 :green_heart: | compile | 0m 10s | the patch passed with JDK Ubuntu-11.0.21+9-post-Ubuntu-0ubuntu122.04 |
| +1 :green_heart: | javac | 0m 10s | the patch passed |
| +1 :green_heart: | compile | 0m 11s | the patch passed with JDK Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| +1 :green_heart: | javac | 0m 11s | the patch passed |
| +1 :green_heart: | checkstyle | 0m 5s | the patch passed |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 6s | the patch passed with JDK Ubuntu-11.0.21+9-post-Ubuntu-0ubuntu122.04 |
| +1 :green_heart: | javadoc | 0m 6s | the patch passed with JDK Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| +1 :green_heart: | findbugs | 0m 23s | the patch passed |
||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 11s | tez-examples in the patch passed. |
| +1 :green_heart: | asflicense | 0m 15s | The patch does not generate ASF License warnings. |
| | | 35m 37s | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.43 ServerAPI=1.43 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/1/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/320 |
| JIRA Issue | TEZ-3268 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 26636412b540 5.15.0-88-generic #98-Ubuntu SMP Mon Oct 2 15:18:56 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / 0c5cf6886 |
| Default Java | Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.21+9-post-Ubuntu-0ubuntu122.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_392-8u392-ga-1~22.04-b08 |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/1/testReport/ |
| Max. process+thread count | 95 (vs. ulimit of 5500) |
| modules | C: tez-examples U: tez-examples |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/1/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "abstractdog (via GitHub)" <gi...@apache.org>.
abstractdog commented on code in PR #320:
URL: https://github.com/apache/tez/pull/320#discussion_r1703083974
##########
tez-examples/src/main/java/org/apache/tez/examples/Demuxer.java:
##########
@@ -0,0 +1,231 @@
+/**
+ * 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.tez.examples;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.tez.client.TezClient;
+import org.apache.tez.common.Preconditions;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.GroupInputEdge;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.ProcessorDescriptor;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager;
+import org.apache.tez.dag.library.vertexmanager.FairShuffleVertexManager.FairRoutingType;
+import org.apache.tez.examples.HashJoinExample.ForwardingProcessor;
+import org.apache.tez.mapreduce.input.MRInput;
+import org.apache.tez.mapreduce.output.MultiMROutput;
+import org.apache.tez.mapreduce.processor.SimpleMRProcessor;
+import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.ProcessorContext;
+import org.apache.tez.runtime.library.api.KeyValueWriterWithBasePath;
+import org.apache.tez.runtime.library.api.KeyValuesReader;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration.ReportPartitionStats;
+import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
+import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValuesInput;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple example of a demuxer tolerant of data skew problems. The example scans the source files,
+ * partitions records by values, and write them in /path/to/{value}-r-{task id}.
+ */
+public class Demuxer extends TezExampleBase {
+ private static final Logger LOG = LoggerFactory.getLogger(Demuxer.class);
+
+ private static final String DEMUXER_OUTPUT = "demuxerOutput";
+
+ public static void main(String[] args) throws Exception {
+ Demuxer job = new Demuxer();
+ int status = ToolRunner.run(new Configuration(), job, args);
+ System.exit(status);
+ }
+
+ @Override
+ protected void printUsage() {
+ System.err.println("Usage: "
+ + "demuxer <inputPath> <outPath> <numPartitions> [isPrecise(default false)]");
+ }
+
+ @Override
+ protected int validateArgs(String[] otherArgs) {
+ if (otherArgs.length < 3 || otherArgs.length > 4) {
+ return 2;
+ }
+ return 0;
+ }
+
+ @Override
+ protected int runJob(String[] args, TezConfiguration tezConf, TezClient tezClient)
+ throws Exception {
+ String inputDirs = args[0];
+ String outputDir = args[1];
+ int numPartitions = Integer.parseInt(args[2]);
+ if (args.length == 4 && Boolean.parseBoolean(args[3])) {
+ tezConf.set(
+ TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS,
+ ReportPartitionStats.PRECISE.getType());
+ }
+
+ List<Path> inputPaths = Arrays
+ .stream(inputDirs.split(","))
+ .map(Path::new)
+ .collect(Collectors.toList());
+ Path outputPath = new Path(outputDir);
+
+ FileSystem fs = outputPath.getFileSystem(tezConf);
+ outputPath = fs.makeQualified(outputPath);
+ if (fs.exists(outputPath)) {
+ System.err.println("Output directory: " + outputDir + " already exists");
+ return 3;
+ }
+ DAG dag = inputPaths.size() == 1
+ ? createDag(tezConf, inputPaths.get(0), outputPath, numPartitions)
+ : createDagWithUnion(tezConf, inputPaths, outputPath, numPartitions);
+ LOG.info("Running Demuxer");
+ return runDag(dag, isCountersLog(), LOG);
+ }
+
+ private DAG createDag(TezConfiguration tezConf, Path inputPath, Path outputPath,
+ int numPartitions) {
+ Vertex inputVertex = createInputVertex(tezConf, "input", inputPath);
+ Vertex demuxVertex = createDemuxVertex(tezConf, outputPath, numPartitions);
+ OrderedPartitionedKVEdgeConfig edgeConf = OrderedPartitionedKVEdgeConfig
+ .newBuilder(
+ Text.class.getName(),
+ NullWritable.class.getName(),
+ HashPartitioner.class.getName())
+ .setFromConfiguration(tezConf)
+ .build();
+ return DAG
+ .create("Demuxer")
+ .addVertex(inputVertex)
+ .addVertex(demuxVertex)
+ .addEdge(Edge.create(inputVertex, demuxVertex, edgeConf.createDefaultEdgeProperty()));
+ }
+
+ private DAG createDagWithUnion(TezConfiguration tezConf, List<Path> inputPaths, Path outputPath,
Review Comment:
https://github.com/apache/tez/pull/306 was closed due to some reasons
is this PR suitable to be adapted to work without that patch?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] TEZ-3268: Provide a demuxer sample app that uses fair routing [tez]
Posted by "tez-yetus (via GitHub)" <gi...@apache.org>.
tez-yetus commented on PR #320:
URL: https://github.com/apache/tez/pull/320#issuecomment-2267538487
:broken_heart: **-1 overall**
| Vote | Subsystem | Runtime | Comment |
|:----:|----------:|--------:|:--------|
| +0 :ok: | reexec | 12m 31s | Docker mode activated. |
||| _ Prechecks _ |
| +1 :green_heart: | dupname | 0m 0s | No case conflicting files found. |
| +1 :green_heart: | @author | 0m 0s | The patch does not contain any @author tags. |
| -1 :x: | test4tests | 0m 0s | The patch doesn't appear to include any new or modified tests. Please justify why no new tests are needed for this patch. Also please list what manual steps were performed to verify this patch. |
||| _ master Compile Tests _ |
| +1 :green_heart: | mvninstall | 16m 21s | master passed |
| +1 :green_heart: | compile | 0m 19s | master passed with JDK Ubuntu-11.0.24+8-post-Ubuntu-1ubuntu322.04 |
| +1 :green_heart: | compile | 0m 18s | master passed with JDK Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| +1 :green_heart: | checkstyle | 1m 9s | master passed |
| +1 :green_heart: | javadoc | 0m 29s | master passed with JDK Ubuntu-11.0.24+8-post-Ubuntu-1ubuntu322.04 |
| +1 :green_heart: | javadoc | 0m 15s | master passed with JDK Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| +0 :ok: | spotbugs | 1m 0s | Used deprecated FindBugs config; considering switching to SpotBugs. |
| +1 :green_heart: | findbugs | 0m 58s | master passed |
||| _ Patch Compile Tests _ |
| +1 :green_heart: | mvninstall | 0m 11s | the patch passed |
| +1 :green_heart: | compile | 0m 10s | the patch passed with JDK Ubuntu-11.0.24+8-post-Ubuntu-1ubuntu322.04 |
| +1 :green_heart: | javac | 0m 11s | the patch passed |
| +1 :green_heart: | compile | 0m 9s | the patch passed with JDK Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| +1 :green_heart: | javac | 0m 9s | the patch passed |
| +1 :green_heart: | checkstyle | 0m 6s | the patch passed |
| +1 :green_heart: | whitespace | 0m 0s | The patch has no whitespace issues. |
| +1 :green_heart: | javadoc | 0m 7s | the patch passed with JDK Ubuntu-11.0.24+8-post-Ubuntu-1ubuntu322.04 |
| +1 :green_heart: | javadoc | 0m 7s | the patch passed with JDK Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| +1 :green_heart: | findbugs | 0m 26s | the patch passed |
||| _ Other Tests _ |
| +1 :green_heart: | unit | 0m 11s | tez-examples in the patch passed. |
| +1 :green_heart: | asflicense | 0m 14s | The patch does not generate ASF License warnings. |
| | | 34m 43s | |
| Subsystem | Report/Notes |
|----------:|:-------------|
| Docker | ClientAPI=1.46 ServerAPI=1.46 base: https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/2/artifact/out/Dockerfile |
| GITHUB PR | https://github.com/apache/tez/pull/320 |
| JIRA Issue | TEZ-3268 |
| Optional Tests | dupname asflicense javac javadoc unit spotbugs findbugs checkstyle compile |
| uname | Linux 3b16b1bdd8b2 5.15.0-117-generic #127-Ubuntu SMP Fri Jul 5 20:13:28 UTC 2024 x86_64 x86_64 x86_64 GNU/Linux |
| Build tool | maven |
| Personality | personality/tez.sh |
| git revision | master / 70d79b27c |
| Default Java | Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| Multi-JDK versions | /usr/lib/jvm/java-11-openjdk-amd64:Ubuntu-11.0.24+8-post-Ubuntu-1ubuntu322.04 /usr/lib/jvm/java-8-openjdk-amd64:Private Build-1.8.0_422-8u422-b05-1~22.04-b05 |
| Test Results | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/2/testReport/ |
| Max. process+thread count | 95 (vs. ulimit of 5500) |
| modules | C: tez-examples U: tez-examples |
| Console output | https://ci-hadoop.apache.org/job/tez-multibranch/job/PR-320/2/console |
| versions | git=2.34.1 maven=3.6.3 findbugs=3.0.1 |
| Powered by | Apache Yetus 0.12.0 https://yetus.apache.org |
This message was automatically generated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@tez.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org