You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by jo...@apache.org on 2018/10/17 01:13:17 UTC
[incubator-nemo] branch tpch-fix created (now 8a7ae95)
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a change to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git.
at 8a7ae95 merge
This branch includes the following new commits:
new 10470a0 tpch
new c6cf3db Merge branch '82-beam-sql-simple' into tpch
new 413ba29 tpch setup
new 68ab8ef q6 success
new 0092a53 all queries run
new 4072654 generic
new a471303 query from file
new a24c55d queryfilepath
new 5ff7062 load only the used tables
new 4e98b4f update
new 60f5067 substring
new 7392dd4 get query string
new e419597 log
new 8a7ae95 merge
The 14 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-nemo] 10/14: update
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 4e98b4f27abd3d9e37c5d36eded542346c003814
Author: John Yang <jo...@apache.org>
AuthorDate: Thu Sep 13 10:26:38 2018 +0900
update
---
.../beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index bf1393c..08ab84e 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -93,9 +93,9 @@ public final class Tpch {
.put("nation", Schemas.NATION_SCHEMA)
.put("region", Schemas.REGION_SCHEMA)
- /*
.put("part", Schemas.PART_SCHEMA)
.put("partsupp", Schemas.PARTSUPP_SCHEMA)
+ /*
.put("store_sales", Schemas.STORE_SALES_SCHEMA)
.put("catalog_sales", Schemas.CATALOG_SALES_SCHEMA)
.put("item", Schemas.ITEM_SCHEMA)
@@ -117,7 +117,7 @@ public final class Tpch {
if (tokens.contains(tableName)) {
LOG.info("HIT: tablename {}", tableName);
- final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
+ final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl*";
final PCollection<Row> table = GenericSourceSink.read(pipeline, filePattern)
.apply("StringToRow", new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat))
.setCoder(tableSchema.getValue().getRowCoder())
[incubator-nemo] 07/14: query from file
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit a471303f5b3a0ee29c5a674849f21eb20d71baca
Author: John Yang <jo...@apache.org>
AuthorDate: Wed Sep 12 17:33:19 2018 +0900
query from file
---
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 56 ++++++++++++++++++++--
.../apache/nemo/examples/beam/SQLTpchITCase.java | 20 ++++++++
2 files changed, 72 insertions(+), 4 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 42db3da..0b62e8e 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -19,7 +19,6 @@ package org.apache.nemo.examples.beam.tpch;
import com.google.common.collect.ImmutableMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.extensions.sql.SqlTransform;
-import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTable;
import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTableProvider;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -35,9 +34,16 @@ import org.apache.nemo.examples.beam.GenericSourceSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.io.Serializable;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beamRow2CsvLine;
@@ -214,7 +220,7 @@ public final class Tpch {
PCollectionTuple tables = PCollectionTuple.empty(pipeline);
for (final Map.Entry<String, Schema> tableSchema : hSchemas.entrySet()) {
- final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl*";
+ final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
final PCollection<Row> table = GenericSourceSink.read(pipeline, filePattern)
.apply("StringToRow", new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat))
.setCoder(tableSchema.getValue().getRowCoder())
@@ -242,7 +248,7 @@ public final class Tpch {
idToQuery.put(5, QUERY5);
idToQuery.put(6, QUERY6);
- LOG.info("{} / {}", queryId, inputDirectory, outputFilePath);
+ LOG.info("{} / {} / {}", queryId, inputDirectory, outputFilePath);
final PipelineOptions options = PipelineOptionsFactory.create().as(NemoPipelineOptions.class);
options.setRunner(NemoPipelineRunner.class);
@@ -257,7 +263,7 @@ public final class Tpch {
final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory);
// Run the TPC-H query
- final PCollection<Row> result = tables.apply(SqlTransform.query(idToQuery.get(queryId)));
+ final PCollection<Row> result = tables.apply(SqlTransform.query(getQueryString(queryId)));
final PCollection<String> resultToWrite = result.apply(MapElements.into(TypeDescriptors.strings()).via(
new SerializableFunction<Row, String>() {
@@ -273,4 +279,46 @@ public final class Tpch {
// Then run
p.run();
}
+
+ private static String getQueryString(final int queryNum) {
+ boolean isStarted = false;
+ final List<String> lines = new ArrayList<>();
+
+
+ final String path =
+ "/Users/johnyang/Documents/workspace/hive-testbench/sample-queries-tpch/tpch_query" + queryNum + ".sql";
+
+ try (final Stream<String> stream = Files.lines(Paths.get(path))) {
+ for (final String line : stream.collect(Collectors.toList())) {
+ if (line.equals("select")) {
+ isStarted = true;
+ }
+
+ if (isStarted) {
+ lines.add(line);
+ }
+ }
+
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ System.out.println(lines);
+ lines.remove(lines.size() - 1);
+
+ final StringBuilder sb = new StringBuilder();
+ lines.forEach(line -> {
+ sb.append(" ");
+ sb.append(line);
+ });
+
+ final String concate = sb.toString();
+ System.out.println(concate);
+ final String cleanOne = concate.replaceAll("\n", " ");
+ System.out.println(cleanOne);
+ final String cleanTwo = cleanOne.replaceAll("\t", " ");
+ System.out.println(cleanTwo);
+
+ return cleanTwo;
+ }
}
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
index 87df9ae..80e9f68 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
@@ -27,6 +27,15 @@ import org.junit.runner.RunWith;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
/**
* Test TPC-H program with JobLauncher.
*/
@@ -58,6 +67,17 @@ public final class SQLTpchITCase {
}
@Test (timeout = TIMEOUT)
+ public void testXX() throws Exception {
+ final int queryNum = 12;
+ JobLauncher.main(builder
+ .addUserMain(Tpch.class.getCanonicalName())
+ .addUserArgs(String.valueOf(queryNum), "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addJobId(SQLTpchITCase.class.getSimpleName())
+ .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
+ .build());
+ }
+
+ @Test (timeout = TIMEOUT)
public void testThree() throws Exception {
final int queryNum = 3;
JobLauncher.main(builder
[incubator-nemo] 12/14: get query string
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 7392dd4e65ab038c7978ae91d50f2ab456ca040d
Author: John Yang <jo...@apache.org>
AuthorDate: Thu Sep 13 11:08:35 2018 +0900
get query string
---
client/src/main/java/org/apache/nemo/client/JobLauncher.java | 1 +
.../main/java/org/apache/nemo/examples/beam/tpch/Tpch.java | 12 +-----------
2 files changed, 2 insertions(+), 11 deletions(-)
diff --git a/client/src/main/java/org/apache/nemo/client/JobLauncher.java b/client/src/main/java/org/apache/nemo/client/JobLauncher.java
index 0ae7dfd..91a6329 100644
--- a/client/src/main/java/org/apache/nemo/client/JobLauncher.java
+++ b/client/src/main/java/org/apache/nemo/client/JobLauncher.java
@@ -151,6 +151,7 @@ public final class JobLauncher {
throw new RuntimeException(e);
} finally {
// Close everything that's left
+
driverRPCServer.shutdown();
driverLauncher.close();
final Optional<Throwable> possibleError = driverLauncher.getStatus().getError();
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 8f7c937..1d6f42c 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -175,24 +175,14 @@ public final class Tpch {
}
private static String getQueryString(final String queryFilePath) {
- boolean isStarted = false;
final List<String> lines = new ArrayList<>();
try (final Stream<String> stream = Files.lines(Paths.get(queryFilePath))) {
- for (final String line : stream.collect(Collectors.toList())) {
- if (line.equals("select")) {
- isStarted = true;
- }
-
- if (isStarted) {
- lines.add(line);
- }
- }
+ stream.forEach(lines::add);
} catch (IOException e) {
throw new RuntimeException(e);
}
System.out.println(lines);
- lines.remove(lines.size() - 1);
final StringBuilder sb = new StringBuilder();
lines.forEach(line -> {
[incubator-nemo] 09/14: load only the used tables
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 5ff7062aee00d5a8ec74935e2d4330da71398127
Author: John Yang <jo...@apache.org>
AuthorDate: Thu Sep 13 10:25:36 2018 +0900
load only the used tables
---
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 159 ++++-----------------
1 file changed, 28 insertions(+), 131 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 16a9297..bf1393c 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -38,10 +38,7 @@ import java.io.IOException;
import java.io.Serializable;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -54,115 +51,6 @@ import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beam
public final class Tpch {
private static final Logger LOG = LoggerFactory.getLogger(Tpch.class.getName());
- public static final String QUERY1 =
- "select\n"
- + "\tl_returnflag,\n"
- + "\tl_linestatus,\n"
- + "\tsum(l_quantity) as sum_qty,\n"
- + "\tsum(l_extendedprice) as sum_base_price,\n"
- + "\tsum(l_extendedprice * (1 - l_discount)) as sum_disc_price,\n"
- + "\tsum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,\n"
- + "\tavg(l_quantity) as avg_qty,\n"
- + "\tavg(l_extendedprice) as avg_price,\n"
- + "\tavg(l_discount) as avg_disc,\n"
- + "\tcount(*) as count_order\n"
- + "from\n"
- + "\tlineitem\n"
- + "where\n"
- + "\tl_shipdate <= date '1998-12-01' - interval '90' day (3)\n"
- + "group by\n"
- + "\tl_returnflag,\n"
- + "\tl_linestatus\n"
- + "order by\n"
- + "\tl_returnflag,\n"
- + "\tl_linestatus limit 10";
-
- public static final String QUERY3 =
- "select\n"
- + "\tl_orderkey,\n"
- + "\tsum(l_extendedprice * (1 - l_discount)) as revenue,\n"
- + "\to_orderdate,\n"
- + "\to_shippriority\n"
- + "from\n"
- + "\tcustomer,\n"
- + "\torders,\n"
- + "\tlineitem\n"
- + "where\n"
- + "\tc_mktsegment = 'BUILDING'\n"
- + "\tand c_custkey = o_custkey\n"
- + "\tand l_orderkey = o_orderkey\n"
- + "\tand o_orderdate < date '1995-03-15'\n"
- + "\tand l_shipdate > date '1995-03-15'\n"
- + "group by\n"
- + "\tl_orderkey,\n"
- + "\to_orderdate,\n"
- + "\to_shippriority\n"
- + "order by\n"
- + "\trevenue desc,\n"
- + "\to_orderdate\n"
- + "limit 10";
-
- public static final String QUERY4 =
- "select\n"
- + "\to_orderpriority,\n"
- + "\tcount(*) as order_count\n"
- + "from\n"
- + "\torders\n"
- + "where\n"
- + "\to_orderdate >= date '1993-07-01'\n"
- + "\tand o_orderdate < date '1993-07-01' + interval '3' month\n"
- + "\tand exists (\n"
- + "\t\tselect\n"
- + "\t\t\t*\n"
- + "\t\tfrom\n"
- + "\t\t\tlineitem\n"
- + "\t\twhere\n"
- + "\t\t\tl_orderkey = o_orderkey\n"
- + "\t\t\tand l_commitdate < l_receiptdate\n"
- + "\t)\n"
- + "group by\n"
- + "\to_orderpriority\n"
- + "order by\n"
- + "\to_orderpriority limit 10";
-
- public static final String QUERY5 =
- "select\n"
- + "\tn_name,\n"
- + "\tsum(l_extendedprice * (1 - l_discount)) as revenue\n"
- + "from\n"
- + "\tcustomer,\n"
- + "\torders,\n"
- + "\tlineitem,\n"
- + "\tsupplier,\n"
- + "\tnation,\n"
- + "\tregion\n"
- + "where\n"
- + "\tc_custkey = o_custkey\n"
- + "\tand l_orderkey = o_orderkey\n"
- + "\tand l_suppkey = s_suppkey\n"
- + "\tand c_nationkey = s_nationkey\n"
- + "\tand s_nationkey = n_nationkey\n"
- + "\tand n_regionkey = r_regionkey\n"
- + "\tand r_name = 'ASIA'\n"
- + "\tand o_orderdate >= date '1994-01-01'\n"
- + "\tand o_orderdate < date '1994-01-01' + interval '1' year\n"
- + "group by\n"
- + "\tn_name\n"
- + "order by\n"
- + "\trevenue desc limit 10";
-
- public static final String QUERY6 =
- "select\n"
- + "\tsum(l_extendedprice * l_discount) as revenue\n"
- + "from\n"
- + "\tlineitem\n"
- + "where\n"
- + "\tl_shipdate >= date '1994-01-01'\n"
- + "\tand l_shipdate < date '1994-01-01' + interval '1' year\n"
- + "\tand l_discount between .06 - 0.01 and .06 + 0.01\n"
- + "\tand l_quantity < 24 limit 10";
-
-
/**
* Private Constructor.
*/
@@ -194,7 +82,8 @@ public final class Tpch {
private static PCollectionTuple getHTables(final Pipeline pipeline,
final CSVFormat csvFormat,
- final String inputDirectory) {
+ final String inputDirectory,
+ final String query) {
final ImmutableMap<String, Schema> hSchemas = ImmutableMap.<String, Schema>builder()
.put("lineitem", Schemas.LINEITEM_SCHEMA)
.put("customer", Schemas.CUSTOMER_SCHEMA)
@@ -220,14 +109,28 @@ public final class Tpch {
PCollectionTuple tables = PCollectionTuple.empty(pipeline);
for (final Map.Entry<String, Schema> tableSchema : hSchemas.entrySet()) {
- final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
- final PCollection<Row> table = GenericSourceSink.read(pipeline, filePattern)
- .apply("StringToRow", new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat))
- .setCoder(tableSchema.getValue().getRowCoder())
- .setName(tableSchema.getKey());
- tables = tables.and(new TupleTag<>(tableSchema.getKey()), table);
-
- LOG.info("FilePattern {} / Tables {}", filePattern, tables);
+ final String tableName = tableSchema.getKey();
+
+ final List<String> tokens = Arrays.asList(query.split(" "));
+ LOG.info("Tokens are {}", tokens);
+
+ if (tokens.contains(tableName)) {
+ LOG.info("HIT: tablename {}", tableName);
+
+ final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
+ final PCollection<Row> table = GenericSourceSink.read(pipeline, filePattern)
+ .apply("StringToRow", new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat))
+ .setCoder(tableSchema.getValue().getRowCoder())
+ .setName(tableSchema.getKey());
+ tables = tables.and(new TupleTag<>(tableSchema.getKey()), table);
+
+ LOG.info("FilePattern {} / Tables {}", filePattern, tables);
+ }
+
+
+
+
+
}
return tables;
}
@@ -241,13 +144,6 @@ public final class Tpch {
final String inputDirectory = args[1];
final String outputFilePath = args[2];
- final Map<Integer, String> idToQuery = new HashMap<>();
- idToQuery.put(1, QUERY1);
- idToQuery.put(3, QUERY3);
- idToQuery.put(4, QUERY4);
- idToQuery.put(5, QUERY5);
- idToQuery.put(6, QUERY6);
-
LOG.info("{} / {} / {}", queryFilePath, inputDirectory, outputFilePath);
final PipelineOptions options = PipelineOptionsFactory.create().as(NemoPipelineOptions.class);
@@ -255,15 +151,16 @@ public final class Tpch {
options.setJobName("TPC-H");
final Pipeline p = Pipeline.create(options);
+ final String queryString = getQueryString(queryFilePath);
// Create tables
final CSVFormat csvFormat = CSVFormat.MYSQL
.withDelimiter('|')
.withNullString("")
.withTrailingDelimiter();
- final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory);
+ final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory, queryString);
// Run the TPC-H query
- final PCollection<Row> result = tables.apply(SqlTransform.query(getQueryString(queryFilePath)));
+ final PCollection<Row> result = tables.apply(SqlTransform.query(queryString));
final PCollection<String> resultToWrite = result.apply(MapElements.into(TypeDescriptors.strings()).via(
new SerializableFunction<Row, String>() {
[incubator-nemo] 08/14: queryfilepath
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit a24c55dcf89982fb37f3b4d3342f3b61b6ee8893
Author: John Yang <jo...@apache.org>
AuthorDate: Thu Sep 13 09:53:11 2018 +0900
queryfilepath
---
.../java/org/apache/nemo/examples/beam/tpch/Tpch.java | 16 +++++-----------
1 file changed, 5 insertions(+), 11 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 0b62e8e..16a9297 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -237,7 +237,7 @@ public final class Tpch {
* @param args arguments.
*/
public static void main(final String[] args) {
- final int queryId = Integer.valueOf(args[0]);
+ final String queryFilePath = args[0];
final String inputDirectory = args[1];
final String outputFilePath = args[2];
@@ -248,7 +248,7 @@ public final class Tpch {
idToQuery.put(5, QUERY5);
idToQuery.put(6, QUERY6);
- LOG.info("{} / {} / {}", queryId, inputDirectory, outputFilePath);
+ LOG.info("{} / {} / {}", queryFilePath, inputDirectory, outputFilePath);
final PipelineOptions options = PipelineOptionsFactory.create().as(NemoPipelineOptions.class);
options.setRunner(NemoPipelineRunner.class);
@@ -263,7 +263,7 @@ public final class Tpch {
final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory);
// Run the TPC-H query
- final PCollection<Row> result = tables.apply(SqlTransform.query(getQueryString(queryId)));
+ final PCollection<Row> result = tables.apply(SqlTransform.query(getQueryString(queryFilePath)));
final PCollection<String> resultToWrite = result.apply(MapElements.into(TypeDescriptors.strings()).via(
new SerializableFunction<Row, String>() {
@@ -280,15 +280,10 @@ public final class Tpch {
p.run();
}
- private static String getQueryString(final int queryNum) {
+ private static String getQueryString(final String queryFilePath) {
boolean isStarted = false;
final List<String> lines = new ArrayList<>();
-
-
- final String path =
- "/Users/johnyang/Documents/workspace/hive-testbench/sample-queries-tpch/tpch_query" + queryNum + ".sql";
-
- try (final Stream<String> stream = Files.lines(Paths.get(path))) {
+ try (final Stream<String> stream = Files.lines(Paths.get(queryFilePath))) {
for (final String line : stream.collect(Collectors.toList())) {
if (line.equals("select")) {
isStarted = true;
@@ -298,7 +293,6 @@ public final class Tpch {
lines.add(line);
}
}
-
} catch (IOException e) {
throw new RuntimeException(e);
}
[incubator-nemo] 03/14: tpch setup
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 413ba2906212e4bc7691ae84f07a7fb6034b7d90
Author: John Yang <jo...@gmail.com>
AuthorDate: Sun Sep 9 11:22:05 2018 +0900
tpch setup
---
.../apache/nemo/examples/beam/tpch/Schemas.java | 372 +++++++++++----------
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 43 +--
...leSumSQLITCase.java => SQLSimpleSumITCase.java} | 4 +-
...mSimpleSumSQLITCase.java => SQLTpchITCase.java} | 20 +-
4 files changed, 228 insertions(+), 211 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java
index b464789..a778abe 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java
@@ -21,14 +21,27 @@ import org.apache.beam.sdk.schemas.Schema;
/**
* A simple SQL application.
- * (Copied/Refined from the example code in the Beam repository)
+ * (Copied and adapted from https://github.com/apache/beam/pull/6240)
*/
public final class Schemas {
- // private static final JavaTypeFactory TYPE_FACTORY =
- // new JavaTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
- // private final ImmutableMap<String, TpcdsTable> tableHMap;
- private final ImmutableMap<String, String> columnPrefixes;
- public static Schema storeSalesSchema =
+ /**
+ * Private.
+ */
+ private Schemas() {
+ }
+
+ public static final ImmutableMap<String, String> COLUMN_PREFIX = ImmutableMap.<String, String>builder()
+ .put("lineitem", "l_")
+ .put("customer", "c_")
+ .put("supplier", "s_")
+ .put("partsupp", "ps_")
+ .put("part", "p_")
+ .put("orders", "o_")
+ .put("nation", "n_")
+ .put("region", "r_")
+ .build();
+
+ public static final Schema STORE_SALES_SCHEMA =
Schema.builder()
.addNullableField("ss_sold_date_sk", Schema.FieldType.INT32)
.addNullableField("ss_sold_time_sk", Schema.FieldType.INT32)
@@ -54,7 +67,8 @@ public final class Schemas {
.addNullableField("ss_net_paid_inc_tax", Schema.FieldType.FLOAT)
.addNullableField("ss_net_profit", Schema.FieldType.FLOAT)
.build();
- public static Schema dateDimSchema =
+
+ public static final Schema DATE_DIM_SCHEMA =
Schema.builder()
.addNullableField("d_date_sk", Schema.FieldType.INT32)
.addNullableField("d_date_id", Schema.FieldType.STRING)
@@ -85,76 +99,79 @@ public final class Schemas {
.addNullableField("d_current_quarter", Schema.FieldType.STRING)
.addNullableField("d_current_year", Schema.FieldType.STRING)
.build();
- public static Schema itemSchema =
+
+ public static final Schema ITEM_SCHEMA =
Schema.builder()
.addNullableField("i_item_sk", Schema.FieldType.INT32)
- .addNullableField("i_item_id", Schema.FieldType.STRING) // .string,
- .addNullableField("i_rec_start_date", Schema.FieldType.DATETIME) // .date,
- .addNullableField("i_rec_end_date", Schema.FieldType.DATETIME) // .date,
- .addNullableField("i_item_desc", Schema.FieldType.STRING) // .string,
- .addNullableField("i_current_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField(
- "i_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("i_brand_id", Schema.FieldType.INT32) // .int,
- .addNullableField("i_brand", Schema.FieldType.STRING) // .string,
- .addNullableField("i_class_id", Schema.FieldType.INT32) // .int,
- .addNullableField("i_class", Schema.FieldType.STRING) // .string,
- .addNullableField("i_category_id", Schema.FieldType.INT32) // .int,
- .addNullableField("i_category", Schema.FieldType.STRING) // .string,
- .addNullableField("i_manufact_id", Schema.FieldType.INT32) // .int,
- .addNullableField("i_manufact", Schema.FieldType.STRING) // .string,
- .addNullableField("i_size", Schema.FieldType.STRING) // .string,
- .addNullableField("i_formulation", Schema.FieldType.STRING) // .string,
- .addNullableField("i_color", Schema.FieldType.STRING) // .string,
- .addNullableField("i_units", Schema.FieldType.STRING) // .string,
- .addNullableField("i_container", Schema.FieldType.STRING) // .string,
- .addNullableField("i_manager_id", Schema.FieldType.INT32) // .int,
- .addNullableField("i_product_name", Schema.FieldType.STRING) // .string),
+ .addNullableField("i_item_id", Schema.FieldType.STRING)
+ .addNullableField("i_rec_start_date", Schema.FieldType.DATETIME)
+ .addNullableField("i_rec_end_date", Schema.FieldType.DATETIME)
+ .addNullableField("i_item_desc", Schema.FieldType.STRING)
+ .addNullableField("i_current_price", Schema.FieldType.FLOAT)
+ .addNullableField("i_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("i_brand_id", Schema.FieldType.INT32)
+ .addNullableField("i_brand", Schema.FieldType.STRING)
+ .addNullableField("i_class_id", Schema.FieldType.INT32)
+ .addNullableField("i_class", Schema.FieldType.STRING)
+ .addNullableField("i_category_id", Schema.FieldType.INT32)
+ .addNullableField("i_category", Schema.FieldType.STRING)
+ .addNullableField("i_manufact_id", Schema.FieldType.INT32)
+ .addNullableField("i_manufact", Schema.FieldType.STRING)
+ .addNullableField("i_size", Schema.FieldType.STRING)
+ .addNullableField("i_formulation", Schema.FieldType.STRING)
+ .addNullableField("i_color", Schema.FieldType.STRING)
+ .addNullableField("i_units", Schema.FieldType.STRING)
+ .addNullableField("i_container", Schema.FieldType.STRING)
+ .addNullableField("i_manager_id", Schema.FieldType.INT32)
+ .addNullableField("i_product_name", Schema.FieldType.STRING)
.build();
- public static Schema inventorySchema =
+
+ public static final Schema INVENTORY_SCHEMA =
Schema.builder()
.addNullableField("inv_date_sk", Schema.FieldType.INT32)
.addNullableField("inv_item_sk", Schema.FieldType.INT32)
.addNullableField("inv_warehouse_sk", Schema.FieldType.INT32)
.addNullableField("inv_quantity_on_hand", Schema.FieldType.INT32)
.build();
- public static Schema catalogSalesSchema =
+
+ public static final Schema CATALOG_SALES_SCHEMA =
Schema.builder()
- .addNullableField("cs_sold_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_sold_time_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_bill_customer_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_bill_cdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_bill_hdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_bill_addr_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_customer_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_cdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_hdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_addr_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_call_center_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_catalog_page_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_ship_mode_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_warehouse_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_item_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_promo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_order_number", Schema.FieldType.INT64) // .long,
- .addNullableField("cs_quantity", Schema.FieldType.INT32) // .int,
- .addNullableField("cs_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_discount_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_coupon_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_ext_ship_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_net_paid", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_net_paid_inc_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_net_paid_inc_ship", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_net_paid_inc_ship_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("cs_net_profit", Schema.FieldType.FLOAT) // .decimal(7,2))
+ .addNullableField("cs_sold_date_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_sold_time_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_date_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_bill_customer_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_bill_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_bill_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_bill_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_customer_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_call_center_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_catalog_page_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_ship_mode_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_warehouse_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_item_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_promo_sk", Schema.FieldType.INT32)
+ .addNullableField("cs_order_number", Schema.FieldType.INT64)
+ .addNullableField("cs_quantity", Schema.FieldType.INT32)
+ .addNullableField("cs_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("cs_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("cs_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_discount_amt", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_tax", Schema.FieldType.FLOAT)
+ .addNullableField("cs_coupon_amt", Schema.FieldType.FLOAT)
+ .addNullableField("cs_ext_ship_cost", Schema.FieldType.FLOAT)
+ .addNullableField("cs_net_paid", Schema.FieldType.FLOAT)
+ .addNullableField("cs_net_paid_inc_tax", Schema.FieldType.FLOAT)
+ .addNullableField("cs_net_paid_inc_ship", Schema.FieldType.FLOAT)
+ .addNullableField("cs_net_paid_inc_ship_tax", Schema.FieldType.FLOAT)
+ .addNullableField("cs_net_profit", Schema.FieldType.FLOAT)
.build();
+
public static final Schema ORDER_SCHEMA =
Schema.builder()
.addInt32Field("o_orderkey")
@@ -167,6 +184,7 @@ public final class Schemas {
.addInt32Field("o_shippriority")
.addStringField("o_comment")
.build();
+
public static final Schema CUSTOMER_SCHEMA =
Schema.builder()
.addInt32Field("c_custkey")
@@ -178,27 +196,29 @@ public final class Schemas {
.addStringField("c_mktsegment")
.addStringField("c_comment")
.build();
- public static Schema getCustomerDsSchema =
+
+ public static final Schema CUSTOMER_DS_SCHEMA =
Schema.builder()
- .addNullableField("c_customer_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_customer_id", Schema.FieldType.STRING) // .string,
- .addNullableField("c_current_cdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_current_hdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_current_addr_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_first_shipto_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_first_sales_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("c_salutation", Schema.FieldType.STRING) // .string,
- .addNullableField("c_first_name", Schema.FieldType.STRING) // .string,
- .addNullableField("c_last_name", Schema.FieldType.STRING) // .string,
- .addNullableField("c_preferred_cust_flag", Schema.FieldType.STRING) // .string,
- .addNullableField("c_birth_day", Schema.FieldType.INT32) // .int,
- .addNullableField("c_birth_month", Schema.FieldType.INT32) // .int,
- .addNullableField("c_birth_year", Schema.FieldType.INT32) // .int,
- .addNullableField("c_birth_country", Schema.FieldType.STRING) // .string,
- .addNullableField("c_login", Schema.FieldType.STRING) // .string,
- .addNullableField("c_email_address", Schema.FieldType.STRING) // .string,
- .addNullableField("c_last_review_date", Schema.FieldType.STRING) // .string)
+ .addNullableField("c_customer_sk", Schema.FieldType.INT32)
+ .addNullableField("c_customer_id", Schema.FieldType.STRING)
+ .addNullableField("c_current_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("c_current_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("c_current_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("c_first_shipto_date_sk", Schema.FieldType.INT32)
+ .addNullableField("c_first_sales_date_sk", Schema.FieldType.INT32)
+ .addNullableField("c_salutation", Schema.FieldType.STRING)
+ .addNullableField("c_first_name", Schema.FieldType.STRING)
+ .addNullableField("c_last_name", Schema.FieldType.STRING)
+ .addNullableField("c_preferred_cust_flag", Schema.FieldType.STRING)
+ .addNullableField("c_birth_day", Schema.FieldType.INT32)
+ .addNullableField("c_birth_month", Schema.FieldType.INT32)
+ .addNullableField("c_birth_year", Schema.FieldType.INT32)
+ .addNullableField("c_birth_country", Schema.FieldType.STRING)
+ .addNullableField("c_login", Schema.FieldType.STRING)
+ .addNullableField("c_email_address", Schema.FieldType.STRING)
+ .addNullableField("c_last_review_date", Schema.FieldType.STRING)
.build();
+
public static final Schema LINEITEM_SCHEMA =
Schema.builder()
.addInt32Field("l_orderkey")
@@ -218,30 +238,34 @@ public final class Schemas {
.addStringField("l_shipmode")
.addStringField("l_comment")
.build();
+
public static final Schema PARTSUPP_SCHEMA =
Schema.builder()
- .addInt32Field("ps_partkey") // identifier
- .addInt32Field("ps_suppkey") // identifier
- .addInt32Field("ps_availqty") // integer
- .addFloatField("ps_supplycost") // decimal
- .addStringField("ps_comment") // variable text, size 199
+ .addInt32Field("ps_partkey")
+ .addInt32Field("ps_suppkey")
+ .addInt32Field("ps_availqty")
+ .addFloatField("ps_supplycost")
+ .addStringField("ps_comment")
.build();
+
public static final Schema REGION_SCHEMA =
Schema.builder()
- .addInt32Field("r_regionkey") // identifier
- .addStringField("r_name") // fixed text, size 25
- .addStringField("r_comment") // variable text, size 152
+ .addInt32Field("r_regionkey")
+ .addStringField("r_name")
+ .addStringField("r_comment")
.build();
+
public static final Schema SUPPLIER_SCHEMA =
Schema.builder()
- .addInt32Field("s_suppkey") // identifier
- .addStringField("s_name") // fixed text, size 25
- .addStringField("s_address") // variable text, size 40
- .addInt32Field("s_nationkey") // identifier
- .addStringField("s_phone") // fixed text, size 15
- .addFloatField("s_acctbal") // decimal
- .addStringField("s_comment") // variable text, size 101
+ .addInt32Field("s_suppkey")
+ .addStringField("s_name")
+ .addStringField("s_address")
+ .addInt32Field("s_nationkey")
+ .addStringField("s_phone")
+ .addFloatField("s_acctbal")
+ .addStringField("s_comment")
.build();
+
public static final Schema PART_SCHEMA =
Schema.builder()
.addInt32Field("p_partkey")
@@ -254,6 +278,7 @@ public final class Schemas {
.addFloatField("p_retailprice")
.addStringField("p_comment")
.build();
+
public static final Schema NATION_SCHEMA =
Schema.builder()
.addInt32Field("n_nationkey")
@@ -261,93 +286,78 @@ public final class Schemas {
.addInt32Field("n_regionkey")
.addStringField("n_comment")
.build();
- public static Schema promotionSchema =
+
+ public static final Schema PROMOTION_SCHEMA =
Schema.builder()
.addNullableField("p_promo_sk", Schema.FieldType.INT32)
- .addNullableField("p_promo_id", Schema.FieldType.STRING) // .string,
- .addNullableField("p_start_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("p_end_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("p_item_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("p_cost", Schema.FieldType.FLOAT) // .decimal(15,2),
- .addNullableField("p_response_target", Schema.FieldType.INT32) // .int,
- .addNullableField("p_promo_name", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_dmail", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_email", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_catalog", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_tv", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_radio", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_press", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_event", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_demo", Schema.FieldType.STRING) // .string,
- .addNullableField("p_channel_details", Schema.FieldType.STRING) // .string,
- .addNullableField("p_purpose", Schema.FieldType.STRING) // .string,
- .addNullableField("p_discount_active", Schema.FieldType.STRING) // .string),
+ .addNullableField("p_promo_id", Schema.FieldType.STRING)
+ .addNullableField("p_start_date_sk", Schema.FieldType.INT32)
+ .addNullableField("p_end_date_sk", Schema.FieldType.INT32)
+ .addNullableField("p_item_sk", Schema.FieldType.INT32)
+ .addNullableField("p_cost", Schema.FieldType.FLOAT)
+ .addNullableField("p_response_target", Schema.FieldType.INT32)
+ .addNullableField("p_promo_name", Schema.FieldType.STRING)
+ .addNullableField("p_channel_dmail", Schema.FieldType.STRING)
+ .addNullableField("p_channel_email", Schema.FieldType.STRING)
+ .addNullableField("p_channel_catalog", Schema.FieldType.STRING)
+ .addNullableField("p_channel_tv", Schema.FieldType.STRING)
+ .addNullableField("p_channel_radio", Schema.FieldType.STRING)
+ .addNullableField("p_channel_press", Schema.FieldType.STRING)
+ .addNullableField("p_channel_event", Schema.FieldType.STRING)
+ .addNullableField("p_channel_demo", Schema.FieldType.STRING)
+ .addNullableField("p_channel_details", Schema.FieldType.STRING)
+ .addNullableField("p_purpose", Schema.FieldType.STRING)
+ .addNullableField("p_discount_active", Schema.FieldType.STRING)
.build();
- public static Schema customerDemographicsSchema =
+
+ public static final Schema CUSTOMER_DEMOGRAPHIC_SCHEMA =
Schema.builder()
.addNullableField("cd_demo_sk", Schema.FieldType.INT32)
- .addNullableField("cd_gender", Schema.FieldType.STRING) // .string,
- .addNullableField("cd_marital_status", Schema.FieldType.STRING) // .string,
- .addNullableField("cd_education_status", Schema.FieldType.STRING) // .string,
- .addNullableField("cd_purchase_estimate", Schema.FieldType.INT32) // .int,
- .addNullableField("cd_credit_rating", Schema.FieldType.STRING) // .string,
- .addNullableField("cd_dep_count", Schema.FieldType.INT32) // .int,
- .addNullableField("cd_dep_employed_count", Schema.FieldType.INT32) // .int,
- .addNullableField("cd_dep_college_count", Schema.FieldType.INT32) // .int),
+ .addNullableField("cd_gender", Schema.FieldType.STRING)
+ .addNullableField("cd_marital_status", Schema.FieldType.STRING)
+ .addNullableField("cd_education_status", Schema.FieldType.STRING)
+ .addNullableField("cd_purchase_estimate", Schema.FieldType.INT32)
+ .addNullableField("cd_credit_rating", Schema.FieldType.STRING)
+ .addNullableField("cd_dep_count", Schema.FieldType.INT32)
+ .addNullableField("cd_dep_employed_count", Schema.FieldType.INT32)
+ .addNullableField("cd_dep_college_count", Schema.FieldType.INT32)
.build();
- public static Schema webSalesSchema =
+
+ public static final Schema WEB_SALES_SCHEMA =
Schema.builder()
- .addNullableField("ws_sold_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_sold_time_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_date_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_item_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_bill_customer_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_bill_cdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_bill_hdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_bill_addr_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_customer_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_cdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_hdemo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_addr_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_web_page_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_web_site_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_ship_mode_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_warehouse_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_promo_sk", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_order_number", Schema.FieldType.INT64) // .long,
- .addNullableField("ws_quantity", Schema.FieldType.INT32) // .int,
- .addNullableField("ws_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_discount_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_coupon_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_ext_ship_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_net_paid", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_net_paid_inc_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_net_paid_inc_ship", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_net_paid_inc_ship_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
- .addNullableField("ws_net_profit", Schema.FieldType.FLOAT) // .decimal(7,2)),
+ .addNullableField("ws_sold_date_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_sold_time_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_date_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_item_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_bill_customer_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_bill_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_bill_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_bill_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_customer_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_web_page_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_web_site_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_ship_mode_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_warehouse_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_promo_sk", Schema.FieldType.INT32)
+ .addNullableField("ws_order_number", Schema.FieldType.INT64)
+ .addNullableField("ws_quantity", Schema.FieldType.INT32)
+ .addNullableField("ws_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("ws_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("ws_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_discount_amt", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_tax", Schema.FieldType.FLOAT)
+ .addNullableField("ws_coupon_amt", Schema.FieldType.FLOAT)
+ .addNullableField("ws_ext_ship_cost", Schema.FieldType.FLOAT)
+ .addNullableField("ws_net_paid", Schema.FieldType.FLOAT)
+ .addNullableField("ws_net_paid_inc_tax", Schema.FieldType.FLOAT)
+ .addNullableField("ws_net_paid_inc_ship", Schema.FieldType.FLOAT)
+ .addNullableField("ws_net_paid_inc_ship_tax", Schema.FieldType.FLOAT)
+ .addNullableField("ws_net_profit", Schema.FieldType.FLOAT)
.build();
- public Schemas() {
- // final ImmutableMap.Builder<String, TpcdsTable> builder = ImmutableMap.builder();
- // for (TpcdsTable<?> tpcdsTable : TpcdsTable.getTables()) {
- // builder.put(tpcdsTable.getTableName(), tpcdsTable);
- // }
- // this.tableHMap = builder.build();
- this.columnPrefixes =
- ImmutableMap.<String, String>builder()
- .put("lineitem", "l_")
- .put("customer", "c_")
- .put("supplier", "s_")
- .put("partsupp", "ps_")
- .put("part", "p_")
- .put("orders", "o_")
- .put("nation", "n_")
- .put("region", "r_")
- .build();
- }
}
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 4ebf0f3..4f45920 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -41,7 +41,7 @@ import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beam
/**
* A simple SQL application.
- * (Copied/Refined from the example code in the Beam repository)
+ * (Copied and adapted from https://github.com/apache/beam/pull/6240)
*/
public final class Tpch {
private static final Logger LOG = LoggerFactory.getLogger(Tpch.class.getName());
@@ -161,11 +161,14 @@ public final class Tpch {
private Tpch() {
}
+ /**
+ * Row csv formats.
+ */
static class RowToCsv extends PTransform<PCollection<Row>, PCollection<String>> implements Serializable {
- private CSVFormat csvFormat;
+ private final CSVFormat csvFormat;
- public RowToCsv(CSVFormat csvFormat) {
+ RowToCsv(final CSVFormat csvFormat) {
this.csvFormat = csvFormat;
}
@@ -174,7 +177,7 @@ public final class Tpch {
}
@Override
- public PCollection<String> expand(PCollection<Row> input) {
+ public PCollection<String> expand(final PCollection<Row> input) {
return input.apply(
"rowToCsv",
MapElements.into(TypeDescriptors.strings()).via(row -> beamRow2CsvLine(row, csvFormat)));
@@ -185,22 +188,24 @@ public final class Tpch {
final CSVFormat csvFormat,
final String inputDirectory) {
final ImmutableMap<String, Schema> hSchemas = ImmutableMap.<String, Schema>builder()
- //.put("customer", Schemas.CUSTOMER_SCHEMA)
.put("lineitem", Schemas.LINEITEM_SCHEMA)
- //.put("nation", Schemas.NATION_SCHEMA)
- //.put("orders", Schemas.ORDER_SCHEMA)
- //.put("part", Schemas.PART_SCHEMA)
- //.put("partsupp", Schemas.PARTSUPP_SCHEMA)
- //.put("region", Schemas.REGION_SCHEMA)
- //.put("supplier", Schemas.SUPPLIER_SCHEMA)
- // .put("store_sales", Schemas.storeSalesSchema)
- // .put("catalog_sales", Schemas.catalogSalesSchema)
- // .put("item", Schemas.itemSchema)
- // .put("date_dim", Schemas.dateDimSchema)
- // .put("promotion", Schemas.promotionSchema)
- // .put("customer_demographics", Schemas.customerDemographicsSchema)
- // .put("web_sales", Schemas.webSalesSchema)
- // .put("inventory", Schemas.inventorySchema)
+ /*
+ .put("customer", Schemas.CUSTOMER_SCHEMA)
+ .put("nation", Schemas.NATION_SCHEMA)
+ .put("orders", Schemas.ORDER_SCHEMA)
+ .put("part", Schemas.PART_SCHEMA)
+ .put("partsupp", Schemas.PARTSUPP_SCHEMA)
+ .put("region", Schemas.REGION_SCHEMA)
+ .put("supplier", Schemas.SUPPLIER_SCHEMA)
+ .put("store_sales", Schemas.STORE_SALES_SCHEMA)
+ .put("catalog_sales", Schemas.CATALOG_SALES_SCHEMA)
+ .put("item", Schemas.ITEM_SCHEMA)
+ .put("date_dim", Schemas.DATE_DIM_SCHEMA)
+ .put("promotion", Schemas.PROMOTION_SCHEMA)
+ .put("customer_demographics", Schemas.CUSTOMER_DEMOGRAPHIC_SCHEMA)
+ .put("web_sales", Schemas.WEB_SALES_SCHEMA)
+ .put("inventory", Schemas.INVENTORY_SCHEMA)
+ */
.build();
PCollectionTuple tables = PCollectionTuple.empty(pipeline);
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLSimpleSumITCase.java
similarity index 95%
copy from examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java
copy to examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLSimpleSumITCase.java
index 4d55ade..a61ed70 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLSimpleSumITCase.java
@@ -31,7 +31,7 @@ import org.powermock.modules.junit4.PowerMockRunner;
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest(JobLauncher.class)
-public final class BeamSimpleSumSQLITCase {
+public final class SQLSimpleSumITCase {
private static final int TIMEOUT = 180000;
private static ArgBuilder builder;
private static final String fileBasePath = System.getProperty("user.dir") + "/../resources/";
@@ -61,7 +61,7 @@ public final class BeamSimpleSumSQLITCase {
@Test (timeout = TIMEOUT)
public void test() throws Exception {
JobLauncher.main(builder
- .addJobId(BeamSimpleSumSQLITCase.class.getSimpleName())
+ .addJobId(SQLSimpleSumITCase.class.getSimpleName())
.addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
.build());
}
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
similarity index 81%
rename from examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java
rename to examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
index 4d55ade..0b7668e 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/BeamSimpleSumSQLITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
@@ -17,8 +17,8 @@ package org.apache.nemo.examples.beam;
import org.apache.nemo.client.JobLauncher;
import org.apache.nemo.common.test.ArgBuilder;
-import org.apache.nemo.common.test.ExampleTestUtil;
import org.apache.nemo.examples.beam.policy.DefaultPolicyParallelismFive;
+import org.apache.nemo.examples.beam.tpch.Tpch;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -27,11 +27,11 @@ import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
/**
- * Test Broadcast program with JobLauncher.
+ * Test TPC-H program with JobLauncher.
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest(JobLauncher.class)
-public final class BeamSimpleSumSQLITCase {
+public final class SQLTpchITCase {
private static final int TIMEOUT = 180000;
private static ArgBuilder builder;
private static final String fileBasePath = System.getProperty("user.dir") + "/../resources/";
@@ -44,25 +44,27 @@ public final class BeamSimpleSumSQLITCase {
@Before
public void setUp() throws Exception {
builder = new ArgBuilder()
- .addUserMain(SimpleSumSQL.class.getCanonicalName())
- .addUserArgs(outputFilePath)
.addResourceJson(executorResourceFileName);
}
@After
public void tearDown() throws Exception {
+ /*
try {
ExampleTestUtil.ensureOutputValidity(fileBasePath, outputFileName, expectedOutputFileName);
} finally {
ExampleTestUtil.deleteOutputFile(fileBasePath, outputFileName);
}
+ */
}
@Test (timeout = TIMEOUT)
- public void test() throws Exception {
+ public void testSix() throws Exception {
JobLauncher.main(builder
- .addJobId(BeamSimpleSumSQLITCase.class.getSimpleName())
- .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
- .build());
+ .addUserMain(Tpch.class.getCanonicalName())
+ .addUserArgs("6", "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addJobId(SQLTpchITCase.class.getSimpleName())
+ .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
+ .build());
}
}
[incubator-nemo] 05/14: all queries run
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 0092a5357c7b9326301a18d386b109d805be5b4b
Author: John Yang <jo...@gmail.com>
AuthorDate: Mon Sep 10 10:01:20 2018 +0900
all queries run
---
.../compiler/frontend/beam/PipelineTranslator.java | 1 +
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 16 +++++-----
.../apache/nemo/examples/beam/SQLTpchITCase.java | 36 +++++++++++++++++++++-
3 files changed, 45 insertions(+), 8 deletions(-)
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
index 67b8828..5d77296 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
@@ -237,6 +237,7 @@ public final class PipelineTranslator
final boolean handlesBeamRow = Stream
.concat(transformVertex.getNode().getInputs().values().stream(),
transformVertex.getNode().getOutputs().values().stream())
+ .filter(pValue -> getCoder(pValue, ctx.pipeline) instanceof KvCoder)
.map(pValue -> (KvCoder) getCoder(pValue, ctx.pipeline)) // Input and output of combine should be KV
.map(kvCoder -> kvCoder.getValueCoder().getEncodedTypeDescriptor()) // We're interested in the 'Value' of KV
.anyMatch(valueTypeDescriptor -> TypeDescriptor.of(Row.class).equals(valueTypeDescriptor));
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index ad93bd9..3ce3072 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -191,14 +191,16 @@ public final class Tpch {
final String inputDirectory) {
final ImmutableMap<String, Schema> hSchemas = ImmutableMap.<String, Schema>builder()
.put("lineitem", Schemas.LINEITEM_SCHEMA)
- /*
.put("customer", Schemas.CUSTOMER_SCHEMA)
- .put("nation", Schemas.NATION_SCHEMA)
.put("orders", Schemas.ORDER_SCHEMA)
+
+ .put("supplier", Schemas.SUPPLIER_SCHEMA)
+ .put("nation", Schemas.NATION_SCHEMA)
+ .put("region", Schemas.REGION_SCHEMA)
+
+ /*
.put("part", Schemas.PART_SCHEMA)
.put("partsupp", Schemas.PARTSUPP_SCHEMA)
- .put("region", Schemas.REGION_SCHEMA)
- .put("supplier", Schemas.SUPPLIER_SCHEMA)
.put("store_sales", Schemas.STORE_SALES_SCHEMA)
.put("catalog_sales", Schemas.CATALOG_SALES_SCHEMA)
.put("item", Schemas.ITEM_SCHEMA)
@@ -248,7 +250,7 @@ public final class Tpch {
final PipelineOptions options = PipelineOptionsFactory.create().as(NemoPipelineOptions.class);
options.setRunner(NemoPipelineRunner.class);
- options.setJobName("SimpleSQL");
+ options.setJobName("TPC-H");
final Pipeline p = Pipeline.create(options);
// Create tables
@@ -265,8 +267,8 @@ public final class Tpch {
new SerializableFunction<Row, String>() {
@Override
public String apply(final Row input) {
- System.out.println("row: " + input.getValues());
- return "row: " + input.getValues();
+ System.out.println(input.getValues().toString());
+ return input.getValues().toString();
}
}));
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
index 427d053..87df9ae 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
@@ -58,10 +58,44 @@ public final class SQLTpchITCase {
}
@Test (timeout = TIMEOUT)
+ public void testThree() throws Exception {
+ final int queryNum = 3;
+ JobLauncher.main(builder
+ .addUserMain(Tpch.class.getCanonicalName())
+ .addUserArgs(String.valueOf(queryNum), "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addJobId(SQLTpchITCase.class.getSimpleName())
+ .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
+ .build());
+ }
+
+ @Test (timeout = TIMEOUT)
+ public void testFour() throws Exception {
+ final int queryNum = 4;
+ JobLauncher.main(builder
+ .addUserMain(Tpch.class.getCanonicalName())
+ .addUserArgs(String.valueOf(queryNum), "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addJobId(SQLTpchITCase.class.getSimpleName())
+ .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
+ .build());
+ }
+
+ @Test (timeout = TIMEOUT)
+ public void testFive() throws Exception {
+ final int queryNum = 5;
+ JobLauncher.main(builder
+ .addUserMain(Tpch.class.getCanonicalName())
+ .addUserArgs(String.valueOf(queryNum), "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addJobId(SQLTpchITCase.class.getSimpleName())
+ .addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
+ .build());
+ }
+
+ @Test (timeout = TIMEOUT)
public void testSix() throws Exception {
+ final int queryNum = 6;
JobLauncher.main(builder
.addUserMain(Tpch.class.getCanonicalName())
- .addUserArgs("6", "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
+ .addUserArgs(String.valueOf(queryNum), "/home/johnyangk/Desktop/tpc-concat-tbls/", outputFilePath)
.addJobId(SQLTpchITCase.class.getSimpleName())
.addOptimizationPolicy(DefaultPolicyParallelismFive.class.getCanonicalName())
.build());
[incubator-nemo] 14/14: merge
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 8a7ae953f3a9de59733df65e27d67a41269431c7
Merge: e419597 18b61c1
Author: John Yang <jo...@apache.org>
AuthorDate: Wed Oct 17 10:12:58 2018 +0900
merge
.../java/org/apache/nemo/common/HashRange.java | 5 +-
.../java/org/apache/nemo/common/StateMachine.java | 21 +-
.../main/java/org/apache/nemo/common/dag/DAG.java | 59 +--
.../main/java/org/apache/nemo/common/dag/Edge.java | 6 +-
.../java/org/apache/nemo/common/dag/Vertex.java | 7 +-
.../org/apache/nemo/common/ir/edge/IREdge.java | 14 +-
.../ir/executionproperty/ExecutionPropertyMap.java | 24 +-
.../org/apache/nemo/common/ir/vertex/IRVertex.java | 15 +-
.../apache/nemo/common/ir/vertex/LoopVertex.java | 43 +-
.../nemo/common/ir/vertex/OperatorVertex.java | 13 +-
compiler/frontend/beam/pom.xml | 5 +
.../compiler/frontend/beam/BeamKeyExtractor.java | 7 +-
.../compiler/frontend/beam/NemoPipelineRunner.java | 5 +-
.../compiler/frontend/beam/PipelineTranslator.java | 141 +++---
.../beam/source/BeamBoundedSourceVertex.java | 46 +-
.../BroadcastVariableSideInputReader.java | 60 +++
.../beam/transform/CreateViewTransform.java | 20 +-
.../beam/transform/DefaultOutputManager.java | 51 +++
.../frontend/beam/transform/DoFnTransform.java | 157 +++++++
.../frontend/beam/transform/DoTransform.java | 490 ---------------------
.../frontend/beam/transform/FlattenTransform.java | 8 +-
.../beam/transform/GroupByKeyTransform.java | 15 +-
.../frontend/beam/transform/WindowFnTransform.java | 98 +++++
.../frontend/beam/transform/WindowTransform.java | 61 ---
.../nemo/compiler/optimizer/PairKeyExtractor.java} | 17 +-
.../compiletime/reshaping/SkewReshapingPass.java | 3 +-
.../compiler/backend/nemo/DAGConverterTest.java | 5 +-
.../frontend/beam/transform/DoFnTransformTest.java | 287 ++++++++++++
.../beam/MultinomialLogisticRegression.java | 6 +-
.../nemo/examples/beam/BeamSimpleSumSQLITCase.java | 68 +++
.../nemo/runtime/common/plan/RuntimeEdge.java | 13 +-
.../org/apache/nemo/runtime/common/plan/Stage.java | 17 +-
.../apache/nemo/runtime/common/plan/StageEdge.java | 19 +-
.../nemo/runtime/common/state/BlockState.java | 4 +-
.../nemo/runtime/common/state/PlanState.java | 4 +-
.../nemo/runtime/common/state/StageState.java | 4 +-
.../nemo/runtime/common/state/TaskState.java | 4 +-
.../master/resource/ExecutorRepresenter.java | 14 +-
.../runtime/master/scheduler/BatchScheduler.java | 4 +-
.../runtime/master/scheduler/TaskRetryTest.java | 4 +-
40 files changed, 1033 insertions(+), 811 deletions(-)
diff --cc compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
index 5d77296,2486a00..729aaad
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
@@@ -237,8 -276,7 +276,8 @@@ public final class PipelineTranslato
final boolean handlesBeamRow = Stream
.concat(transformVertex.getNode().getInputs().values().stream(),
transformVertex.getNode().getOutputs().values().stream())
- .filter(pValue -> getCoder(pValue, ctx.pipeline) instanceof KvCoder)
- .map(pValue -> (KvCoder) getCoder(pValue, ctx.pipeline)) // Input and output of combine should be KV
++ .filter(pValue -> getCoder(pValue, ctx.root) instanceof KvCoder)
+ .map(pValue -> (KvCoder) getCoder(pValue, ctx.root)) // Input and output of combine should be KV
.map(kvCoder -> kvCoder.getValueCoder().getEncodedTypeDescriptor()) // We're interested in the 'Value' of KV
.anyMatch(valueTypeDescriptor -> TypeDescriptor.of(Row.class).equals(valueTypeDescriptor));
if (handlesBeamRow) {
@@@ -490,17 -529,18 +530,17 @@@
+ "for an edge from %s to %s", communicationPatternSelector, src, dst));
}
final IREdge edge = new IREdge(communicationPattern, src, dst);
- final Coder<?> coder;
+ final Coder coder;
+ final Coder windowCoder;
if (input instanceof PCollection) {
coder = ((PCollection) input).getCoder();
+ windowCoder = ((PCollection) input).getWindowingStrategy().getWindowFn().windowCoder();
} else if (input instanceof PCollectionView) {
- coder = getCoderForView((PCollectionView) input, pipeline);
+ coder = getCoderForView((PCollectionView) input, root);
- windowCoder = ((PCollectionView) input).getPCollection()
- .getWindowingStrategy().getWindowFn().windowCoder();
++ windowCoder = ((PCollectionView) input).getPCollection().getWindowingStrategy().getWindowFn().windowCoder();
} else {
- coder = null;
- }
- if (coder == null) {
throw new RuntimeException(String.format("While adding an edge from %s, to %s, coder for PValue %s cannot "
- + "be determined", src, dst, input));
+ + "be determined", src, dst, input));
}
edge.setProperty(KeyExtractorProperty.of(new BeamKeyExtractor()));
[incubator-nemo] 13/14: log
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit e419597e0e3e07b38af643c6ea0e86a759fa810d
Author: John Yang <jo...@apache.org>
AuthorDate: Sat Sep 29 10:43:49 2018 +0900
log
---
.../apache/nemo/compiler/frontend/beam/transform/DoTransform.java | 5 +++++
.../beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java | 2 +-
2 files changed, 6 insertions(+), 1 deletion(-)
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
index 1935370..11f0bd8 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
@@ -35,6 +35,8 @@ import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TupleTag;
import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
@@ -49,6 +51,7 @@ import java.util.Map;
* @param <O> output type.
*/
public final class DoTransform<I, O> implements Transform<I, O> {
+ private static final Logger LOG = LoggerFactory.getLogger(DoTransform.class.getName());
private final DoFn doFn;
private final ObjectMapper mapper;
private final String serializedOptions;
@@ -214,12 +217,14 @@ public final class DoTransform<I, O> implements Transform<I, O> {
@Override
public void write(final Integer state) {
+ LOG.info("WRITE STATE {} to {}", state, idToInteger);
idToInteger.put(stateId, state);
}
@Nullable
@Override
public Integer read() {
+ LOG.info("READ STATE {} from {}", stateId, idToInteger);
return idToInteger.get(stateId);
}
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 1d6f42c..f770a7a 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -39,7 +39,6 @@ import java.io.Serializable;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beamRow2CsvLine;
@@ -163,6 +162,7 @@ public final class Tpch {
new SerializableFunction<Row, String>() {
@Override
public String apply(final Row input) {
+ System.out.println("ROW INPUT VISITED");
System.out.println(input.getValues().toString());
return input.getValues().toString();
}
[incubator-nemo] 11/14: substring
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 60f50674c83d4b9f7aad4732bc61090e94aa237a
Author: John Yang <jo...@apache.org>
AuthorDate: Thu Sep 13 10:32:08 2018 +0900
substring
---
.../beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java | 5 +----
1 file changed, 1 insertion(+), 4 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 08ab84e..8f7c937 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -111,10 +111,7 @@ public final class Tpch {
for (final Map.Entry<String, Schema> tableSchema : hSchemas.entrySet()) {
final String tableName = tableSchema.getKey();
- final List<String> tokens = Arrays.asList(query.split(" "));
- LOG.info("Tokens are {}", tokens);
-
- if (tokens.contains(tableName)) {
+ if (query.contains(tableName)) {
LOG.info("HIT: tablename {}", tableName);
final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl*";
[incubator-nemo] 04/14: q6 success
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 68ab8efc4464aa3d77f91943bb91ee0f3ddde90e
Author: John Yang <jo...@gmail.com>
AuthorDate: Sun Sep 9 18:57:01 2018 +0900
q6 success
---
.../compiler/frontend/beam/PipelineTranslator.java | 12 +++++++
.../frontend/beam/transform/DoTransform.java | 38 +++++++++++++++++++++-
.../nemo/examples/beam/GenericSourceSink.java | 2 +-
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 11 ++++---
.../apache/nemo/examples/beam/SQLTpchITCase.java | 7 ++--
5 files changed, 60 insertions(+), 10 deletions(-)
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
index 6ef96bf..67b8828 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineTranslator.java
@@ -117,6 +117,18 @@ public final class PipelineTranslator
private static void parDoSingleOutputTranslator(final TranslationContext ctx,
final PrimitiveTransformVertex transformVertex,
final ParDo.SingleOutput<?, ?> transform) {
+ /*
+ final DoFnSignature signature = DoFnSignatures.getSignature(transform.getFn().getClass());
+ final Map<String, DoFnSignature.StateDeclaration> stateDeclarationMap = signature.stateDeclarations();
+ for (final DoFnSignature.StateDeclaration declaration : stateDeclarationMap.values()) {
+ final TypeDescriptor<? extends State> stateType = declaration.stateType();
+ if (stateType.isSubtypeOf(TypeDescriptor.of(ValueState.class))) {
+ declaration.field()
+ } else {
+ throw new UnsupportedOperationException("Only ValueState is supported at the moment");
+ }
+ }
+ */
final DoTransform doTransform = new DoTransform(transform.getFn(), ctx.pipelineOptions);
final IRVertex vertex = new OperatorVertex(doTransform);
ctx.addVertex(vertex);
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
index 3a36ec0..1935370 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/transform/DoTransform.java
@@ -16,6 +16,7 @@
package org.apache.nemo.compiler.frontend.beam.transform;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.beam.sdk.state.ValueState;
import org.apache.nemo.common.ir.OutputCollector;
import org.apache.nemo.common.ir.vertex.transform.Transform;
import org.apache.nemo.runtime.executor.datatransfer.OutputCollectorImpl;
@@ -35,7 +36,9 @@ import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TupleTag;
import org.joda.time.Instant;
+import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -200,6 +203,37 @@ public final class DoTransform<I, O> implements Transform<I, O> {
private final ObjectMapper mapper;
private final PipelineOptions options;
+ private final Map<String, TmpIntegerValueState> idToValueState;
+ private final Map<String, Integer> idToInteger;
+ private class TmpIntegerValueState implements ValueState<Integer> {
+ final String stateId;
+
+ public TmpIntegerValueState(final String stateId) {
+ this.stateId = stateId;
+ }
+
+ @Override
+ public void write(final Integer state) {
+ idToInteger.put(stateId, state);
+ }
+
+ @Nullable
+ @Override
+ public Integer read() {
+ return idToInteger.get(stateId);
+ }
+
+ @Override
+ public ValueState<Integer> readLater() {
+ return this;
+ }
+
+ @Override
+ public void clear() {
+ // Do nothing.
+ }
+ }
+
/**
* ProcessContext Constructor.
*
@@ -222,6 +256,8 @@ public final class DoTransform<I, O> implements Transform<I, O> {
} catch (IOException e) {
throw new RuntimeException(e);
}
+ this.idToValueState = new HashMap<>(1);
+ this.idToInteger = new HashMap<>(1);
}
/**
@@ -369,7 +405,7 @@ public final class DoTransform<I, O> implements Transform<I, O> {
@Override
public State state(final String stateId) {
- throw new UnsupportedOperationException("state() in ProcessContext under DoTransform");
+ return idToValueState.computeIfAbsent(stateId, TmpIntegerValueState::new);
}
@Override
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/GenericSourceSink.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/GenericSourceSink.java
index 51fd3bd..5ac8b20 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/GenericSourceSink.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/GenericSourceSink.java
@@ -39,7 +39,7 @@ import org.slf4j.LoggerFactory;
* Helper class for handling source/sink in a generic way.
* Assumes String-type PCollections.
*/
-final class GenericSourceSink {
+public final class GenericSourceSink {
/**
* Default Constructor.
*/
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 4f45920..ad93bd9 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -26,10 +26,12 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.*;
import org.apache.commons.csv.CSVFormat;
import org.apache.nemo.compiler.frontend.beam.NemoPipelineOptions;
import org.apache.nemo.compiler.frontend.beam.NemoPipelineRunner;
+import org.apache.nemo.examples.beam.GenericSourceSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -250,24 +252,25 @@ public final class Tpch {
final Pipeline p = Pipeline.create(options);
// Create tables
- final CSVFormat csvFormat = CSVFormat.MYSQL.withDelimiter('|').withNullString("");
+ final CSVFormat csvFormat = CSVFormat.MYSQL
+ .withDelimiter('|')
+ .withNullString("")
+ .withTrailingDelimiter();
final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory);
// Run the TPC-H query
final PCollection<Row> result = tables.apply(SqlTransform.query(idToQuery.get(queryId)));
- /*
final PCollection<String> resultToWrite = result.apply(MapElements.into(TypeDescriptors.strings()).via(
new SerializableFunction<Row, String>() {
@Override
- public String apply(Row input) {
+ public String apply(final Row input) {
System.out.println("row: " + input.getValues());
return "row: " + input.getValues();
}
}));
GenericSourceSink.write(resultToWrite, outputFilePath);
- */
// Then run
p.run();
diff --git a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
index 0b7668e..427d053 100644
--- a/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
+++ b/examples/beam/src/test/java/org/apache/nemo/examples/beam/SQLTpchITCase.java
@@ -17,6 +17,7 @@ package org.apache.nemo.examples.beam;
import org.apache.nemo.client.JobLauncher;
import org.apache.nemo.common.test.ArgBuilder;
+import org.apache.nemo.common.test.ExampleTestUtil;
import org.apache.nemo.examples.beam.policy.DefaultPolicyParallelismFive;
import org.apache.nemo.examples.beam.tpch.Tpch;
import org.junit.After;
@@ -36,8 +37,8 @@ public final class SQLTpchITCase {
private static ArgBuilder builder;
private static final String fileBasePath = System.getProperty("user.dir") + "/../resources/";
- private static final String outputFileName = "test_output_simplesql";
- private static final String expectedOutputFileName = "expected_output_simplesql";
+ private static final String outputFileName = "test_output_tpch";
+ private static final String expectedOutputFileName = "expected_output_tpch";
private static final String executorResourceFileName = fileBasePath + "beam_test_executor_resources.json";
private static final String outputFilePath = fileBasePath + outputFileName;
@@ -49,13 +50,11 @@ public final class SQLTpchITCase {
@After
public void tearDown() throws Exception {
- /*
try {
ExampleTestUtil.ensureOutputValidity(fileBasePath, outputFileName, expectedOutputFileName);
} finally {
ExampleTestUtil.deleteOutputFile(fileBasePath, outputFileName);
}
- */
}
@Test (timeout = TIMEOUT)
[incubator-nemo] 06/14: generic
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 4072654d07390410ead7d77a138e480c98032dfd
Author: John Yang <jo...@apache.org>
AuthorDate: Wed Sep 12 10:48:21 2018 +0900
generic
---
.../src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java | 10 +++-------
1 file changed, 3 insertions(+), 7 deletions(-)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
index 3ce3072..42db3da 100644
--- a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -214,13 +214,9 @@ public final class Tpch {
PCollectionTuple tables = PCollectionTuple.empty(pipeline);
for (final Map.Entry<String, Schema> tableSchema : hSchemas.entrySet()) {
- final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
- final PCollection<Row> table = new TextTable(
- tableSchema.getValue(),
- filePattern,
- new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat),
- new RowToCsv(csvFormat))
- .buildIOReader(pipeline.begin())
+ final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl*";
+ final PCollection<Row> table = GenericSourceSink.read(pipeline, filePattern)
+ .apply("StringToRow", new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat))
.setCoder(tableSchema.getValue().getRowCoder())
.setName(tableSchema.getKey());
tables = tables.and(new TupleTag<>(tableSchema.getKey()), table);
[incubator-nemo] 01/14: tpch
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 10470a09793bf8b999a3cf129b55d8fbf141f176
Author: John Yang <jo...@gmail.com>
AuthorDate: Sun Sep 9 10:34:15 2018 +0900
tpch
---
.../apache/nemo/examples/beam/tpch/Schemas.java | 353 +++++++++++++++++++++
.../org/apache/nemo/examples/beam/tpch/Tpch.java | 270 ++++++++++++++++
2 files changed, 623 insertions(+)
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java
new file mode 100644
index 0000000..b464789
--- /dev/null
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Schemas.java
@@ -0,0 +1,353 @@
+/*
+ * 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.nemo.examples.beam.tpch;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.beam.sdk.schemas.Schema;
+
+/**
+ * A simple SQL application.
+ * (Copied/Refined from the example code in the Beam repository)
+ */
+public final class Schemas {
+ // private static final JavaTypeFactory TYPE_FACTORY =
+ // new JavaTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
+ // private final ImmutableMap<String, TpcdsTable> tableHMap;
+ private final ImmutableMap<String, String> columnPrefixes;
+ public static Schema storeSalesSchema =
+ Schema.builder()
+ .addNullableField("ss_sold_date_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_sold_time_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_item_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_customer_sk", Schema.FieldType.STRING)
+ .addNullableField("ss_cdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_hdemo_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_addr_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_store_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_promo_sk", Schema.FieldType.INT32)
+ .addNullableField("ss_ticket_number", Schema.FieldType.INT64)
+ .addNullableField("ss_quantity", Schema.FieldType.INT32)
+ .addNullableField("ss_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("ss_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("ss_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("ss_ext_discount_amt", Schema.FieldType.FLOAT)
+ .addNullableField("ss_ext_sales_price", Schema.FieldType.FLOAT)
+ .addNullableField("ss_ext_wholesale_cost", Schema.FieldType.FLOAT)
+ .addNullableField("ss_ext_list_price", Schema.FieldType.FLOAT)
+ .addNullableField("ss_ext_tax", Schema.FieldType.FLOAT)
+ .addNullableField("ss_coupon_amt", Schema.FieldType.FLOAT)
+ .addNullableField("ss_net_paid", Schema.FieldType.FLOAT)
+ .addNullableField("ss_net_paid_inc_tax", Schema.FieldType.FLOAT)
+ .addNullableField("ss_net_profit", Schema.FieldType.FLOAT)
+ .build();
+ public static Schema dateDimSchema =
+ Schema.builder()
+ .addNullableField("d_date_sk", Schema.FieldType.INT32)
+ .addNullableField("d_date_id", Schema.FieldType.STRING)
+ .addNullableField("d_date", Schema.FieldType.STRING)
+ .addNullableField("d_month_seq", Schema.FieldType.INT32)
+ .addNullableField("d_week_seq", Schema.FieldType.INT32)
+ .addNullableField("d_quarter_seq", Schema.FieldType.INT32)
+ .addNullableField("d_year", Schema.FieldType.INT32)
+ .addNullableField("d_dow", Schema.FieldType.INT32)
+ .addNullableField("d_moy", Schema.FieldType.INT32)
+ .addNullableField("d_dom", Schema.FieldType.INT32)
+ .addNullableField("d_qoy", Schema.FieldType.INT32)
+ .addNullableField("d_fy_year", Schema.FieldType.INT32)
+ .addNullableField("d_fy_quarter_seq", Schema.FieldType.INT32)
+ .addNullableField("d_fy_week_seq", Schema.FieldType.INT32)
+ .addNullableField("d_day_name", Schema.FieldType.STRING)
+ .addNullableField("d_quarter_name", Schema.FieldType.STRING)
+ .addNullableField("d_holiday", Schema.FieldType.STRING)
+ .addNullableField("d_weekend", Schema.FieldType.STRING)
+ .addNullableField("d_following_holiday", Schema.FieldType.STRING)
+ .addNullableField("d_first_dom", Schema.FieldType.INT32)
+ .addNullableField("d_last_dom", Schema.FieldType.INT32)
+ .addNullableField("d_same_day_ly", Schema.FieldType.INT32)
+ .addNullableField("d_same_day_lq", Schema.FieldType.INT32)
+ .addNullableField("d_current_day", Schema.FieldType.STRING)
+ .addNullableField("d_current_week", Schema.FieldType.STRING)
+ .addNullableField("d_current_month", Schema.FieldType.STRING)
+ .addNullableField("d_current_quarter", Schema.FieldType.STRING)
+ .addNullableField("d_current_year", Schema.FieldType.STRING)
+ .build();
+ public static Schema itemSchema =
+ Schema.builder()
+ .addNullableField("i_item_sk", Schema.FieldType.INT32)
+ .addNullableField("i_item_id", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_rec_start_date", Schema.FieldType.DATETIME) // .date,
+ .addNullableField("i_rec_end_date", Schema.FieldType.DATETIME) // .date,
+ .addNullableField("i_item_desc", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_current_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField(
+ "i_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("i_brand_id", Schema.FieldType.INT32) // .int,
+ .addNullableField("i_brand", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_class_id", Schema.FieldType.INT32) // .int,
+ .addNullableField("i_class", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_category_id", Schema.FieldType.INT32) // .int,
+ .addNullableField("i_category", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_manufact_id", Schema.FieldType.INT32) // .int,
+ .addNullableField("i_manufact", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_size", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_formulation", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_color", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_units", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_container", Schema.FieldType.STRING) // .string,
+ .addNullableField("i_manager_id", Schema.FieldType.INT32) // .int,
+ .addNullableField("i_product_name", Schema.FieldType.STRING) // .string),
+ .build();
+ public static Schema inventorySchema =
+ Schema.builder()
+ .addNullableField("inv_date_sk", Schema.FieldType.INT32)
+ .addNullableField("inv_item_sk", Schema.FieldType.INT32)
+ .addNullableField("inv_warehouse_sk", Schema.FieldType.INT32)
+ .addNullableField("inv_quantity_on_hand", Schema.FieldType.INT32)
+ .build();
+ public static Schema catalogSalesSchema =
+ Schema.builder()
+ .addNullableField("cs_sold_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_sold_time_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_bill_customer_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_bill_cdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_bill_hdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_bill_addr_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_customer_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_cdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_hdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_addr_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_call_center_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_catalog_page_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_ship_mode_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_warehouse_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_item_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_promo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_order_number", Schema.FieldType.INT64) // .long,
+ .addNullableField("cs_quantity", Schema.FieldType.INT32) // .int,
+ .addNullableField("cs_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_discount_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_coupon_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_ext_ship_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_net_paid", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_net_paid_inc_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_net_paid_inc_ship", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_net_paid_inc_ship_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("cs_net_profit", Schema.FieldType.FLOAT) // .decimal(7,2))
+ .build();
+ public static final Schema ORDER_SCHEMA =
+ Schema.builder()
+ .addInt32Field("o_orderkey")
+ .addInt32Field("o_custkey")
+ .addStringField("o_orderstatus")
+ .addFloatField("o_totalprice")
+ .addStringField("o_orderdate")
+ .addStringField("o_orderpriority")
+ .addStringField("o_clerk")
+ .addInt32Field("o_shippriority")
+ .addStringField("o_comment")
+ .build();
+ public static final Schema CUSTOMER_SCHEMA =
+ Schema.builder()
+ .addInt32Field("c_custkey")
+ .addStringField("c_name")
+ .addStringField("c_address")
+ .addInt32Field("c_nationkey")
+ .addStringField("c_phone")
+ .addFloatField("c_acctbal")
+ .addStringField("c_mktsegment")
+ .addStringField("c_comment")
+ .build();
+ public static Schema getCustomerDsSchema =
+ Schema.builder()
+ .addNullableField("c_customer_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_customer_id", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_current_cdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_current_hdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_current_addr_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_first_shipto_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_first_sales_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_salutation", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_first_name", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_last_name", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_preferred_cust_flag", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_birth_day", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_birth_month", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_birth_year", Schema.FieldType.INT32) // .int,
+ .addNullableField("c_birth_country", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_login", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_email_address", Schema.FieldType.STRING) // .string,
+ .addNullableField("c_last_review_date", Schema.FieldType.STRING) // .string)
+ .build();
+ public static final Schema LINEITEM_SCHEMA =
+ Schema.builder()
+ .addInt32Field("l_orderkey")
+ .addInt32Field("l_partkey")
+ .addInt32Field("l_suppkey")
+ .addInt32Field("l_linenumber")
+ .addFloatField("l_quantity")
+ .addFloatField("l_extendedprice")
+ .addFloatField("l_discount")
+ .addFloatField("l_tax")
+ .addStringField("l_returnflag")
+ .addStringField("l_linestatus")
+ .addStringField("l_shipdate")
+ .addStringField("l_commitdate")
+ .addStringField("l_receiptdate")
+ .addStringField("l_shipinstruct")
+ .addStringField("l_shipmode")
+ .addStringField("l_comment")
+ .build();
+ public static final Schema PARTSUPP_SCHEMA =
+ Schema.builder()
+ .addInt32Field("ps_partkey") // identifier
+ .addInt32Field("ps_suppkey") // identifier
+ .addInt32Field("ps_availqty") // integer
+ .addFloatField("ps_supplycost") // decimal
+ .addStringField("ps_comment") // variable text, size 199
+ .build();
+ public static final Schema REGION_SCHEMA =
+ Schema.builder()
+ .addInt32Field("r_regionkey") // identifier
+ .addStringField("r_name") // fixed text, size 25
+ .addStringField("r_comment") // variable text, size 152
+ .build();
+ public static final Schema SUPPLIER_SCHEMA =
+ Schema.builder()
+ .addInt32Field("s_suppkey") // identifier
+ .addStringField("s_name") // fixed text, size 25
+ .addStringField("s_address") // variable text, size 40
+ .addInt32Field("s_nationkey") // identifier
+ .addStringField("s_phone") // fixed text, size 15
+ .addFloatField("s_acctbal") // decimal
+ .addStringField("s_comment") // variable text, size 101
+ .build();
+ public static final Schema PART_SCHEMA =
+ Schema.builder()
+ .addInt32Field("p_partkey")
+ .addStringField("p_name")
+ .addStringField("p_mfgr")
+ .addStringField("p_brand")
+ .addStringField("p_type")
+ .addInt32Field("p_size")
+ .addStringField("p_container")
+ .addFloatField("p_retailprice")
+ .addStringField("p_comment")
+ .build();
+ public static final Schema NATION_SCHEMA =
+ Schema.builder()
+ .addInt32Field("n_nationkey")
+ .addStringField("n_name")
+ .addInt32Field("n_regionkey")
+ .addStringField("n_comment")
+ .build();
+ public static Schema promotionSchema =
+ Schema.builder()
+ .addNullableField("p_promo_sk", Schema.FieldType.INT32)
+ .addNullableField("p_promo_id", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_start_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("p_end_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("p_item_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("p_cost", Schema.FieldType.FLOAT) // .decimal(15,2),
+ .addNullableField("p_response_target", Schema.FieldType.INT32) // .int,
+ .addNullableField("p_promo_name", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_dmail", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_email", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_catalog", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_tv", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_radio", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_press", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_event", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_demo", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_channel_details", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_purpose", Schema.FieldType.STRING) // .string,
+ .addNullableField("p_discount_active", Schema.FieldType.STRING) // .string),
+ .build();
+ public static Schema customerDemographicsSchema =
+ Schema.builder()
+ .addNullableField("cd_demo_sk", Schema.FieldType.INT32)
+ .addNullableField("cd_gender", Schema.FieldType.STRING) // .string,
+ .addNullableField("cd_marital_status", Schema.FieldType.STRING) // .string,
+ .addNullableField("cd_education_status", Schema.FieldType.STRING) // .string,
+ .addNullableField("cd_purchase_estimate", Schema.FieldType.INT32) // .int,
+ .addNullableField("cd_credit_rating", Schema.FieldType.STRING) // .string,
+ .addNullableField("cd_dep_count", Schema.FieldType.INT32) // .int,
+ .addNullableField("cd_dep_employed_count", Schema.FieldType.INT32) // .int,
+ .addNullableField("cd_dep_college_count", Schema.FieldType.INT32) // .int),
+ .build();
+ public static Schema webSalesSchema =
+ Schema.builder()
+ .addNullableField("ws_sold_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_sold_time_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_date_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_item_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_bill_customer_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_bill_cdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_bill_hdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_bill_addr_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_customer_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_cdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_hdemo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_addr_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_web_page_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_web_site_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_ship_mode_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_warehouse_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_promo_sk", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_order_number", Schema.FieldType.INT64) // .long,
+ .addNullableField("ws_quantity", Schema.FieldType.INT32) // .int,
+ .addNullableField("ws_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_discount_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_sales_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_wholesale_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_list_price", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_coupon_amt", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_ext_ship_cost", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_net_paid", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_net_paid_inc_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_net_paid_inc_ship", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_net_paid_inc_ship_tax", Schema.FieldType.FLOAT) // .decimal(7,2),
+ .addNullableField("ws_net_profit", Schema.FieldType.FLOAT) // .decimal(7,2)),
+ .build();
+ public Schemas() {
+ // final ImmutableMap.Builder<String, TpcdsTable> builder = ImmutableMap.builder();
+ // for (TpcdsTable<?> tpcdsTable : TpcdsTable.getTables()) {
+ // builder.put(tpcdsTable.getTableName(), tpcdsTable);
+ // }
+ // this.tableHMap = builder.build();
+ this.columnPrefixes =
+ ImmutableMap.<String, String>builder()
+ .put("lineitem", "l_")
+ .put("customer", "c_")
+ .put("supplier", "s_")
+ .put("partsupp", "ps_")
+ .put("part", "p_")
+ .put("orders", "o_")
+ .put("nation", "n_")
+ .put("region", "r_")
+ .build();
+ }
+}
diff --git a/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
new file mode 100644
index 0000000..4ebf0f3
--- /dev/null
+++ b/examples/beam/src/main/java/org/apache/nemo/examples/beam/tpch/Tpch.java
@@ -0,0 +1,270 @@
+/*
+ * 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.nemo.examples.beam.tpch;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.SqlTransform;
+import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTable;
+import org.apache.beam.sdk.extensions.sql.meta.provider.text.TextTableProvider;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.*;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.nemo.compiler.frontend.beam.NemoPipelineOptions;
+import org.apache.nemo.compiler.frontend.beam.NemoPipelineRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.beam.sdk.extensions.sql.impl.schema.BeamTableUtils.beamRow2CsvLine;
+
+/**
+ * A simple SQL application.
+ * (Copied/Refined from the example code in the Beam repository)
+ */
+public final class Tpch {
+ private static final Logger LOG = LoggerFactory.getLogger(Tpch.class.getName());
+
+ public static final String QUERY1 =
+ "select\n"
+ + "\tl_returnflag,\n"
+ + "\tl_linestatus,\n"
+ + "\tsum(l_quantity) as sum_qty,\n"
+ + "\tsum(l_extendedprice) as sum_base_price,\n"
+ + "\tsum(l_extendedprice * (1 - l_discount)) as sum_disc_price,\n"
+ + "\tsum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,\n"
+ + "\tavg(l_quantity) as avg_qty,\n"
+ + "\tavg(l_extendedprice) as avg_price,\n"
+ + "\tavg(l_discount) as avg_disc,\n"
+ + "\tcount(*) as count_order\n"
+ + "from\n"
+ + "\tlineitem\n"
+ + "where\n"
+ + "\tl_shipdate <= date '1998-12-01' - interval '90' day (3)\n"
+ + "group by\n"
+ + "\tl_returnflag,\n"
+ + "\tl_linestatus\n"
+ + "order by\n"
+ + "\tl_returnflag,\n"
+ + "\tl_linestatus limit 10";
+
+ public static final String QUERY3 =
+ "select\n"
+ + "\tl_orderkey,\n"
+ + "\tsum(l_extendedprice * (1 - l_discount)) as revenue,\n"
+ + "\to_orderdate,\n"
+ + "\to_shippriority\n"
+ + "from\n"
+ + "\tcustomer,\n"
+ + "\torders,\n"
+ + "\tlineitem\n"
+ + "where\n"
+ + "\tc_mktsegment = 'BUILDING'\n"
+ + "\tand c_custkey = o_custkey\n"
+ + "\tand l_orderkey = o_orderkey\n"
+ + "\tand o_orderdate < date '1995-03-15'\n"
+ + "\tand l_shipdate > date '1995-03-15'\n"
+ + "group by\n"
+ + "\tl_orderkey,\n"
+ + "\to_orderdate,\n"
+ + "\to_shippriority\n"
+ + "order by\n"
+ + "\trevenue desc,\n"
+ + "\to_orderdate\n"
+ + "limit 10";
+
+ public static final String QUERY4 =
+ "select\n"
+ + "\to_orderpriority,\n"
+ + "\tcount(*) as order_count\n"
+ + "from\n"
+ + "\torders\n"
+ + "where\n"
+ + "\to_orderdate >= date '1993-07-01'\n"
+ + "\tand o_orderdate < date '1993-07-01' + interval '3' month\n"
+ + "\tand exists (\n"
+ + "\t\tselect\n"
+ + "\t\t\t*\n"
+ + "\t\tfrom\n"
+ + "\t\t\tlineitem\n"
+ + "\t\twhere\n"
+ + "\t\t\tl_orderkey = o_orderkey\n"
+ + "\t\t\tand l_commitdate < l_receiptdate\n"
+ + "\t)\n"
+ + "group by\n"
+ + "\to_orderpriority\n"
+ + "order by\n"
+ + "\to_orderpriority limit 10";
+
+ public static final String QUERY5 =
+ "select\n"
+ + "\tn_name,\n"
+ + "\tsum(l_extendedprice * (1 - l_discount)) as revenue\n"
+ + "from\n"
+ + "\tcustomer,\n"
+ + "\torders,\n"
+ + "\tlineitem,\n"
+ + "\tsupplier,\n"
+ + "\tnation,\n"
+ + "\tregion\n"
+ + "where\n"
+ + "\tc_custkey = o_custkey\n"
+ + "\tand l_orderkey = o_orderkey\n"
+ + "\tand l_suppkey = s_suppkey\n"
+ + "\tand c_nationkey = s_nationkey\n"
+ + "\tand s_nationkey = n_nationkey\n"
+ + "\tand n_regionkey = r_regionkey\n"
+ + "\tand r_name = 'ASIA'\n"
+ + "\tand o_orderdate >= date '1994-01-01'\n"
+ + "\tand o_orderdate < date '1994-01-01' + interval '1' year\n"
+ + "group by\n"
+ + "\tn_name\n"
+ + "order by\n"
+ + "\trevenue desc limit 10";
+
+ public static final String QUERY6 =
+ "select\n"
+ + "\tsum(l_extendedprice * l_discount) as revenue\n"
+ + "from\n"
+ + "\tlineitem\n"
+ + "where\n"
+ + "\tl_shipdate >= date '1994-01-01'\n"
+ + "\tand l_shipdate < date '1994-01-01' + interval '1' year\n"
+ + "\tand l_discount between .06 - 0.01 and .06 + 0.01\n"
+ + "\tand l_quantity < 24 limit 10";
+
+
+ /**
+ * Private Constructor.
+ */
+ private Tpch() {
+ }
+
+ static class RowToCsv extends PTransform<PCollection<Row>, PCollection<String>> implements Serializable {
+
+ private CSVFormat csvFormat;
+
+ public RowToCsv(CSVFormat csvFormat) {
+ this.csvFormat = csvFormat;
+ }
+
+ public CSVFormat getCsvFormat() {
+ return csvFormat;
+ }
+
+ @Override
+ public PCollection<String> expand(PCollection<Row> input) {
+ return input.apply(
+ "rowToCsv",
+ MapElements.into(TypeDescriptors.strings()).via(row -> beamRow2CsvLine(row, csvFormat)));
+ }
+ }
+
+ private static PCollectionTuple getHTables(final Pipeline pipeline,
+ final CSVFormat csvFormat,
+ final String inputDirectory) {
+ final ImmutableMap<String, Schema> hSchemas = ImmutableMap.<String, Schema>builder()
+ //.put("customer", Schemas.CUSTOMER_SCHEMA)
+ .put("lineitem", Schemas.LINEITEM_SCHEMA)
+ //.put("nation", Schemas.NATION_SCHEMA)
+ //.put("orders", Schemas.ORDER_SCHEMA)
+ //.put("part", Schemas.PART_SCHEMA)
+ //.put("partsupp", Schemas.PARTSUPP_SCHEMA)
+ //.put("region", Schemas.REGION_SCHEMA)
+ //.put("supplier", Schemas.SUPPLIER_SCHEMA)
+ // .put("store_sales", Schemas.storeSalesSchema)
+ // .put("catalog_sales", Schemas.catalogSalesSchema)
+ // .put("item", Schemas.itemSchema)
+ // .put("date_dim", Schemas.dateDimSchema)
+ // .put("promotion", Schemas.promotionSchema)
+ // .put("customer_demographics", Schemas.customerDemographicsSchema)
+ // .put("web_sales", Schemas.webSalesSchema)
+ // .put("inventory", Schemas.inventorySchema)
+ .build();
+
+ PCollectionTuple tables = PCollectionTuple.empty(pipeline);
+ for (final Map.Entry<String, Schema> tableSchema : hSchemas.entrySet()) {
+ final String filePattern = inputDirectory + tableSchema.getKey() + ".tbl";
+ final PCollection<Row> table = new TextTable(
+ tableSchema.getValue(),
+ filePattern,
+ new TextTableProvider.CsvToRow(tableSchema.getValue(), csvFormat),
+ new RowToCsv(csvFormat))
+ .buildIOReader(pipeline.begin())
+ .setCoder(tableSchema.getValue().getRowCoder())
+ .setName(tableSchema.getKey());
+ tables = tables.and(new TupleTag<>(tableSchema.getKey()), table);
+
+ LOG.info("FilePattern {} / Tables {}", filePattern, tables);
+ }
+ return tables;
+ }
+
+
+ /**
+ * @param args arguments.
+ */
+ public static void main(final String[] args) {
+ final int queryId = Integer.valueOf(args[0]);
+ final String inputDirectory = args[1];
+ final String outputFilePath = args[2];
+
+ final Map<Integer, String> idToQuery = new HashMap<>();
+ idToQuery.put(1, QUERY1);
+ idToQuery.put(3, QUERY3);
+ idToQuery.put(4, QUERY4);
+ idToQuery.put(5, QUERY5);
+ idToQuery.put(6, QUERY6);
+
+ LOG.info("{} / {}", queryId, inputDirectory, outputFilePath);
+
+ final PipelineOptions options = PipelineOptionsFactory.create().as(NemoPipelineOptions.class);
+ options.setRunner(NemoPipelineRunner.class);
+ options.setJobName("SimpleSQL");
+ final Pipeline p = Pipeline.create(options);
+
+ // Create tables
+ final CSVFormat csvFormat = CSVFormat.MYSQL.withDelimiter('|').withNullString("");
+ final PCollectionTuple tables = getHTables(p, csvFormat, inputDirectory);
+
+ // Run the TPC-H query
+ final PCollection<Row> result = tables.apply(SqlTransform.query(idToQuery.get(queryId)));
+
+ /*
+ final PCollection<String> resultToWrite = result.apply(MapElements.into(TypeDescriptors.strings()).via(
+ new SerializableFunction<Row, String>() {
+ @Override
+ public String apply(Row input) {
+ System.out.println("row: " + input.getValues());
+ return "row: " + input.getValues();
+ }
+ }));
+
+ GenericSourceSink.write(resultToWrite, outputFilePath);
+ */
+
+ // Then run
+ p.run();
+ }
+}
[incubator-nemo] 02/14: Merge branch '82-beam-sql-simple' into tpch
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch tpch-fix
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit c6cf3dbeafae69ab2bde22bdae28312070edfbb6
Merge: 10470a0 848ac5b
Author: John Yang <jo...@gmail.com>
AuthorDate: Sun Sep 9 10:35:50 2018 +0900
Merge branch '82-beam-sql-simple' into tpch
.../compiler/frontend/beam/NemoPipelineResult.java | 11 ++-
.../compiler/frontend/beam/PipelineTranslator.java | 107 +++++++++++++--------
examples/beam/pom.xml | 5 +
.../apache/nemo/examples/beam/SimpleSumSQL.java | 89 +++++++++++++++++
.../nemo/examples/beam/BeamSimpleSumSQLITCase.java | 68 +++++++++++++
examples/resources/expected_output_simplesql | 1 +
6 files changed, 239 insertions(+), 42 deletions(-)