You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by dukechain <gi...@git.apache.org> on 2014/11/10 17:30:58 UTC
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
GitHub user dukechain opened a pull request:
https://github.com/apache/incubator-flink/pull/194
[FLINK-1040] Make type() call in projections optional (or remove it)
The original version of projection requires a type() to specify the types of output dataset, e.g., dataset.project(0).type(string.class)). Now, I implement a new interface projection() which does not require the type() call, i.e., dataset.projection(0). Similarly, I modify the operation of JoinOperator and CrossOperator. In particular, some changes in ProjectorOperator.java, JoinOperator.java and CrossOperator.java are generated by TupleGenerator.java.
I add more testing cases here. In addition, I rewrite a new example of web log analysis by the projection() API.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/dukechain/incubator-flink flink-1040
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-flink/pull/194.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #194
----
commit efc6fa30bb6a9cb7fe839586e6781993d45a10fd
Author: dukechain <ch...@gmail.com>
Date: 2014-11-10T15:11:07Z
[FLINK-1040] Make type() call in projections optional (or remove it)
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20137228
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -163,6 +162,31 @@ public DefaultCross(DataSet<I1> input1, DataSet<I2> input2) {
public CrossProjection<I1, I2> projectFirst(int... firstFieldIndexes) {
return new CrossProjection<I1, I2>(input1, input2, firstFieldIndexes, null);
}
+
+
+ /**
+ * Initiates a ProjectCross transformation and projects the first cross input<br/>
+ * If the first cross input is a {@link Tuple} {@link DataSet}, fields can be selected by their index.
+ * If the first cross input is not a Tuple DataSet, no parameters should be passed.<br/>
+ *
+ * Fields of the first and second input can be added by chaining the method calls of
+ * {@link org.apache.flink.api.java.operators.CrossOperator.ProjectCross#projectionFirst(int...)} and
+ * {@link org.apache.flink.api.java.operators.CrossOperator.ProjectCross#projectionSecond(int...)}.
+ *
+ * @param firstFieldIndexes If the first input is a Tuple DataSet, the indexes of the selected fields.
+ * For a non-Tuple DataSet, do not provide parameters.
+ * The order of fields in the output tuple is defined by to the order of field indexes.
+ * @return A CrossProjection to complete the Cross transformation.
+ *
+ * @see Tuple
+ * @see DataSet
+ * @see org.apache.flink.api.java.operators.CrossOperator.CrossProjection
+ * @see org.apache.flink.api.java.operators.CrossOperator.ProjectCross
+ */
+ public <OUT extends Tuple> ProjectCross<I1, I2, OUT> projectionFirst(int... firstFieldIndexes) {
--- End diff --
Same here. No need for two operators that do the same thing. Just replace the {{projectFirst()}} and {{projectSecond()}} by your changes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20138018
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java ---
@@ -1215,8 +1735,7 @@ public JoinProjection(DataSet<I1> ds1, DataSet<I2> ds2, Keys<I1> keys1, Keys<I2>
* Projects a pair of joined elements to a {@link Tuple} with the previously selected fields.
* Requires the classes of the fields of the resulting tuples.
*
- * @param type0 The class of field '0' of the result tuples.
- * @param type1 The class of field '1' of the result tuples.
+ * @param type0 The class of field '0' of the result tuples. @param type1 The class of field '1' of the result tuples.
--- End diff --
add linebreak
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-62520098
@StephanEwen Damn, I was too slow ;-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20142494
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -93,6 +113,30 @@ public Projection(DataSet<T> ds, int[] fieldIndexes) {
this.fieldIndexes = fieldIndexes;
}
+ public void acceptAdditionalIndexes(int... additionalIndexes) {
--- End diff --
Sure! ```acceptAdditionalIndexes()``` should be invisible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20569372
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java ---
@@ -521,7 +521,7 @@ public void join(IN1 first, IN2 second, Collector<OUT> out) throws Exception {
out.collect(function.join(first, second));
}
}
- */
+ */
--- End diff --
trivial: tab
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20569395
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -356,7 +381,7 @@ public CrossProjection(DataSet<I1> ds1, DataSet<I2> ds2, int[] firstFieldIndexes
// check field indexes and adapt to position in tuple
int maxFieldIndex = firstInput ? numFieldsDs1 : numFieldsDs2;
for(int i=0; i<this.fieldIndexes.length; i++) {
- if(this.fieldIndexes[i] > maxFieldIndex - 1) {
+ if(this.fieldIndexes[i] > maxFieldIndex - 1 || this.fieldIndexes[i] < 0) {
--- End diff --
minor: I think we should replace these checks with [Preconditions.checkElementIndex](http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/base/Preconditions.html#checkElementIndex%28int,%20int%29), which is more concise and throws the respective Exception for you (it might be a matter of taste though :P).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-flink/pull/194
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20137955
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java ---
@@ -542,6 +542,33 @@ public void join(IN1 first, IN2 second, Collector<OUT> out) throws Exception {
return new JoinProjection<I1, I2>(getInput1(), getInput2(), getKeys1(), getKeys2(), getJoinHint(), firstFieldIndexes, null);
}
+
+ /**
+ * Initiates a ProjectJoin transformation and projects the first join input<br/>
+ * If the first join input is a {@link Tuple} {@link DataSet}, fields can be selected by their index.
+ * If the first join input is not a Tuple DataSet, no parameters should be passed.<br/>
+ *
+ * Fields of the first and second input can be added by chaining the method calls of
+ * {@link org.apache.flink.api.java.operators.JoinOperator.ProjectJoin#projectionFirst(int...)} and
+ * {@link org.apache.flink.api.java.operators.JoinOperator.ProjectJoin#projectionSecond(int...)}.
+ *
+ * @param firstFieldIndexes If the first input is a Tuple DataSet, the indexes of the selected fields.
+ * For a non-Tuple DataSet, do not provide parameters.
+ * The order of fields in the output tuple is defined by to the order of field indexes.
+ * @return A ProjectJoin to complete the Join transformation.
+ *
+ * @see Tuple
+ * @see DataSet
+ * @see org.apache.flink.api.java.operators.JoinOperator.JoinProjection
+ * @see org.apache.flink.api.java.operators.JoinOperator.ProjectJoin
+ */
+ public <OUT extends Tuple> ProjectJoin<I1, I2, OUT> projectionFirst(int... firstFieldIndexes) {
--- End diff --
rename to ``projectFirst()`` and ``projectSecond()``.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20138116
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -63,11 +73,21 @@ public ProjectOperator(DataSet<IN> input, int[] fields, TupleTypeInfo<OUT> retur
return ppo;
}
-
+
+ @SuppressWarnings("hiding")
+ public <OUT extends Tuple> ProjectOperator<?, OUT> projection(int... fieldIndexes) {
+ proj.acceptAdditionalIndexes(fieldIndexes);
+
+ return proj.types();
+ }
+
+
public static class Projection<T> {
private final DataSet<T> ds;
- private final int[] fieldIndexes;
+// private final int[] fieldIndexes;
--- End diff --
remove line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20704942
--- Diff: flink-compiler/src/test/java/org/apache/flink/compiler/custompartition/CustomPartitioningGlobalOptimizationTest.java ---
@@ -49,10 +52,11 @@ public void testJoinReduceCombination() {
DataSet<Tuple2<Long, Long>> input1 = env.fromElements(new Tuple2<Long, Long>(0L, 0L));
DataSet<Tuple3<Long, Long, Long>> input2 = env.fromElements(new Tuple3<Long, Long, Long>(0L, 0L, 0L));
- DataSet<Tuple3<Long, Long, Long>> joined = input1.join(input2)
+ ProjectJoin<Tuple2<Long, Long>,Tuple3<Long, Long, Long>,Tuple3<Long, Long, Long>> projectjoin = input1.join(input2)
.where(1).equalTo(0)
- .projectFirst(0,1).projectSecond(2).types(Long.class, Long.class, Long.class)
- .withPartitioner(partitioner);
+ .projectFirst(0,1).projectSecond(2);
+
+ DataSet<Tuple3<Long, Long, Long>> joined = projectjoin.withPartitioner(partitioner);
--- End diff --
We should open a JIRA for this issue.
It will confuse users if it is not possible to chain these methods.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20140092
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/WebLogAnalysisWithoutType.java ---
@@ -0,0 +1,334 @@
+/*
+ * 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.flink.examples.java.relational;
+
+
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.util.Collector;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.examples.java.relational.util.WebLogData;
+import org.apache.flink.examples.java.relational.util.WebLogDataGenerator;
+
+/**
+ * This program processes web logs and relational data.
+ * It implements the following relational query:
+ *
+ * <code><pre>
+ * SELECT
+ * r.pageURL,
+ * r.pageRank,
+ * r.avgDuration
+ * FROM documents d JOIN rankings r
+ * ON d.url = r.url
+ * WHERE CONTAINS(d.text, [keywords])
+ * AND r.rank > [rank]
+ * AND NOT EXISTS
+ * (
+ * SELECT * FROM Visits v
+ * WHERE v.destUrl = d.url
+ * AND v.visitDate < [date]
+ * );
+ * </pre></code>
+ *
+ * <p>
+ * Input files are plain text CSV files using the pipe character ('|') as field separator.
+ * The tables referenced in the query can be generated using the {@link WebLogDataGenerator} and
+ * have the following schemas
+ * <code><pre>
+ * CREATE TABLE Documents (
+ * url VARCHAR(100) PRIMARY KEY,
+ * contents TEXT );
+ *
+ * CREATE TABLE Rankings (
+ * pageRank INT,
+ * pageURL VARCHAR(100) PRIMARY KEY,
+ * avgDuration INT );
+ *
+ * CREATE TABLE Visits (
+ * sourceIP VARCHAR(16),
+ * destURL VARCHAR(100),
+ * visitDate DATE,
+ * adRevenue FLOAT,
+ * userAgent VARCHAR(64),
+ * countryCode VARCHAR(3),
+ * languageCode VARCHAR(6),
+ * searchWord VARCHAR(32),
+ * duration INT );
+ * </pre></code>
+ *
+ * <p>
+ * Usage: <code>WebLogAnalysis <documents path> <ranks path> <visits path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WebLogData}.
+ *
+ * <p>
+ * This example shows how to use:
+ * <ul>
+ * <li> tuple data types
+ * <li> projection and join projection
+ * <li> the CoGroup transformation for an anti-join
+ * </ul>
+ *
+ */
+@SuppressWarnings("serial")
+public class WebLogAnalysisWithoutType {
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(String[] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // get input data
+ DataSet<Tuple2<String, String>> documents = getDocumentsDataSet(env);
+ DataSet<Tuple3<Integer, String, Integer>> ranks = getRanksDataSet(env);
+ DataSet<Tuple2<String, String>> visits = getVisitsDataSet(env);
+
+ // Retain documents with keywords
+ /*DataSet<Tuple1<String>> filterDocs = documents
+ .filter(new FilterDocByKeyWords())
+ .projection(0); */
--- End diff --
Sure, just put your changes in an additional commit and push it to your PR branch. The PR will be automatically updated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20136932
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/WebLogAnalysisWithoutType.java ---
@@ -0,0 +1,334 @@
+/*
+ * 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.flink.examples.java.relational;
+
+
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.util.Collector;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.examples.java.relational.util.WebLogData;
+import org.apache.flink.examples.java.relational.util.WebLogDataGenerator;
+
+/**
+ * This program processes web logs and relational data.
+ * It implements the following relational query:
+ *
+ * <code><pre>
+ * SELECT
+ * r.pageURL,
+ * r.pageRank,
+ * r.avgDuration
+ * FROM documents d JOIN rankings r
+ * ON d.url = r.url
+ * WHERE CONTAINS(d.text, [keywords])
+ * AND r.rank > [rank]
+ * AND NOT EXISTS
+ * (
+ * SELECT * FROM Visits v
+ * WHERE v.destUrl = d.url
+ * AND v.visitDate < [date]
+ * );
+ * </pre></code>
+ *
+ * <p>
+ * Input files are plain text CSV files using the pipe character ('|') as field separator.
+ * The tables referenced in the query can be generated using the {@link WebLogDataGenerator} and
+ * have the following schemas
+ * <code><pre>
+ * CREATE TABLE Documents (
+ * url VARCHAR(100) PRIMARY KEY,
+ * contents TEXT );
+ *
+ * CREATE TABLE Rankings (
+ * pageRank INT,
+ * pageURL VARCHAR(100) PRIMARY KEY,
+ * avgDuration INT );
+ *
+ * CREATE TABLE Visits (
+ * sourceIP VARCHAR(16),
+ * destURL VARCHAR(100),
+ * visitDate DATE,
+ * adRevenue FLOAT,
+ * userAgent VARCHAR(64),
+ * countryCode VARCHAR(3),
+ * languageCode VARCHAR(6),
+ * searchWord VARCHAR(32),
+ * duration INT );
+ * </pre></code>
+ *
+ * <p>
+ * Usage: <code>WebLogAnalysis <documents path> <ranks path> <visits path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WebLogData}.
+ *
+ * <p>
+ * This example shows how to use:
+ * <ul>
+ * <li> tuple data types
+ * <li> projection and join projection
+ * <li> the CoGroup transformation for an anti-join
+ * </ul>
+ *
+ */
+@SuppressWarnings("serial")
+public class WebLogAnalysisWithoutType {
--- End diff --
I would replace the current WebLogAnalysis by this one.
We want to keep the number of examples small and there is not reason to keep the old one, IMO.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-62519985
Hi @dukechain,
thanks a lot for this contribution! Removing the need for the ``types()`` call is really great!
As commented inline, I would replace the current API for the projection operator by your implementation, instead of adding new API methods. With your solution, the old project API becomes obsolete. I know, this is an API breaking change, but users only need to remove the types call. Maybe, we can even add a dummy ``types()`` method to the ``ProjectOperator`` and mark it as deprecated.
Porting the API, would also require to adapt all examples and test cases that use projection.
Flink's documentation is also contained in the git repository as .md files. We did this to keep documenation and code in sync. The adapted documenation should be part of an API changing PR. You find the documenation in the ./docs folder. The following files need to be adapted:
- ./docs/programming_guide.md
- ./docs/dataset_transformations.md
Again, great work and thanks for your contribution!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20942004
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java ---
@@ -146,9 +146,9 @@ public boolean filter(Tuple4<Integer, Double, Double, String> l) {
DataSet<Tuple2<Integer, Double>> revenueByCustomer =
ordersFilteredByYear.joinWithHuge(lineitemsFilteredByFlag)
.where(0).equalTo(0)
- .projectFirst(1).projectSecond(1)
- .types(Integer.class, Double.class)
--- End diff --
I would remove them..
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20139821
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/DataSet.java ---
@@ -246,6 +247,24 @@ public ExecutionEnvironment getExecutionEnvironment() {
return new Projection<T>(this, fieldIndexes);
}
+ /**
+ * Initiates a Project transformation on a {@link Tuple} {@link DataSet}.<br/>
+ * <b>Note: Only Tuple DataSets can be projected.</b></br>
+ * The transformation projects each Tuple of the DataSet onto a (sub)set of fields.</br>
+ * This method returns a {@link ProjectOperator} to complete the transformation.
+ *
+ * @param fieldIndexes The field indexes of the input tuples that are retained.
+ * The order of fields in the output tuple corresponds to the order of field indexes.
+ * @return A ProjectOperator to complete the Project transformation.
+ *
+ * @see Tuple
+ * @see DataSet
+ * @see org.apache.flink.api.java.operators.ProjectOperator
+ */
+ public <OUT extends Tuple> ProjectOperator<?, OUT> projection(int... fieldIndexes) {
--- End diff --
I agree with you absolutely. For a safe consideration, I retain the original project() method. I would like to change this API totally.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20137355
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -203,13 +251,40 @@ public DefaultCross(DataSet<I1> input1, DataSet<I2> input2) {
* @see Tuple
* @see DataSet
*/
- private static final class ProjectCross<I1, I2, OUT extends Tuple> extends CrossOperator<I1, I2, OUT> {
+// private static final class ProjectCross<I1, I2, OUT extends Tuple> extends CrossOperator<I1, I2, OUT> {
--- End diff --
remove line
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20141144
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -93,6 +113,30 @@ public Projection(DataSet<T> ds, int[] fieldIndexes) {
this.fieldIndexes = fieldIndexes;
}
+ public void acceptAdditionalIndexes(int... additionalIndexes) {
--- End diff --
In order to remove types() call in project(), project() should return a ProjectOperator object. As projectOperator inherites from DataSet, Hence, at user-end, there is no compiling error for chained projection like dataset.project(0).project(1). If we do not accept the additional index, the run-time result would be wrong.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20141214
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java ---
@@ -323,9 +534,12 @@ private static void modifyJoinProjectOperator(File root) throws IOException {
sb.append("\t\t * Projects a pair of joined elements to a {@link Tuple} with the previously selected fields. \n");
sb.append("\t\t * Requires the classes of the fields of the resulting tuples. \n");
sb.append("\t\t * \n");
+ sb.append("\t\t *");
for (int i = 0; i < numFields; i++) {
- sb.append("\t\t * @param type" + i + " The class of field '"+i+"' of the result tuples.\n");
+ sb.append(" @param type" + i + " The class of field '"+i+"' of the result tuples. ");
--- End diff --
In order to pass "mvn verify", I change it here. Is there a limitation in verification setting (2500 lines)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20137458
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -203,13 +251,40 @@ public DefaultCross(DataSet<I1> input1, DataSet<I2> input2) {
* @see Tuple
* @see DataSet
*/
- private static final class ProjectCross<I1, I2, OUT extends Tuple> extends CrossOperator<I1, I2, OUT> {
+// private static final class ProjectCross<I1, I2, OUT extends Tuple> extends CrossOperator<I1, I2, OUT> {
+ public static final class ProjectCross<I1, I2, OUT extends Tuple> extends CrossOperator<I1, I2, OUT> {
+
+ private CrossProjection<I1, I2> crossProjection;
protected ProjectCross(DataSet<I1> input1, DataSet<I2> input2, int[] fields, boolean[] isFromFirst, TupleTypeInfo<OUT> returnType) {
super(input1, input2,
new ProjectCrossFunction<I1, I2, OUT>(fields, isFromFirst, returnType.createSerializer().createInstance()), returnType);
+
+ crossProjection = null;
+ }
+
+ protected ProjectCross(DataSet<I1> input1, DataSet<I2> input2, int[] fields, boolean[] isFromFirst, TupleTypeInfo<OUT> returnType, CrossProjection<I1, I2> crossProjection) {
+ super(input1, input2,
+ new ProjectCrossFunction<I1, I2, OUT>(fields, isFromFirst, returnType.createSerializer().createInstance()), returnType);
+
+ this.crossProjection = crossProjection;
+ }
+
+ @SuppressWarnings("hiding")
+ public <OUT extends Tuple> ProjectCross<I1, I2, OUT> projectionFirst(int... firstFieldIndexes) {
--- End diff --
rename to ``projectFirst()`` and ``projectSecond()``
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20705922
--- Diff: flink-compiler/src/test/java/org/apache/flink/compiler/custompartition/CustomPartitioningGlobalOptimizationTest.java ---
@@ -49,10 +52,11 @@ public void testJoinReduceCombination() {
DataSet<Tuple2<Long, Long>> input1 = env.fromElements(new Tuple2<Long, Long>(0L, 0L));
DataSet<Tuple3<Long, Long, Long>> input2 = env.fromElements(new Tuple3<Long, Long, Long>(0L, 0L, 0L));
- DataSet<Tuple3<Long, Long, Long>> joined = input1.join(input2)
+ ProjectJoin<Tuple2<Long, Long>,Tuple3<Long, Long, Long>,Tuple3<Long, Long, Long>> projectjoin = input1.join(input2)
.where(1).equalTo(0)
- .projectFirst(0,1).projectSecond(2).types(Long.class, Long.class, Long.class)
- .withPartitioner(partitioner);
+ .projectFirst(0,1).projectSecond(2);
+
+ DataSet<Tuple3<Long, Long, Long>> joined = projectjoin.withPartitioner(partitioner);
--- End diff --
Yes, we can open a JIRA issue after this branch merged.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-64664110
I think it is impossible to catch these mismatches at compilation time if we remove type().
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20572732
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -517,859 +542,426 @@ public CrossProjection(DataSet<I1> ds1, DataSet<I2> ds2, int[] firstFieldIndexes
// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
/**
+ * Chooses a projectTupleX according to the length of {@link CrossProjection#fieldIndexes}
+ *
+ * @return The projected DataSet.
+ *
+ * @see Projection
+ */
+ @SuppressWarnings("unchecked")
+ public <OUT extends Tuple> ProjectCross<I1, I2, OUT> projectTupleX() {
+ ProjectCross<I1, I2, OUT> projectionCross = null;
+
+ switch (fieldIndexes.length) {
+ case 1: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple1(); break;
--- End diff --
Oh, sorry. I didn't see that these are generated anyways...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20140135
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java ---
@@ -1215,8 +1735,7 @@ public JoinProjection(DataSet<I1> ds1, DataSet<I2> ds2, Keys<I1> keys1, Keys<I2>
* Projects a pair of joined elements to a {@link Tuple} with the previously selected fields.
* Requires the classes of the fields of the resulting tuples.
*
- * @param type0 The class of field '0' of the result tuples.
- * @param type1 The class of field '1' of the result tuples.
+ * @param type0 The class of field '0' of the result tuples. @param type1 The class of field '1' of the result tuples.
--- End diff --
It seems "mvn verify" does not permit a .java file exceeds 2500 lines. Anyway, these types() would not be used again if we drop types() in project(). I would remove them totally.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20928980
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java ---
@@ -146,9 +146,9 @@ public boolean filter(Tuple4<Integer, Double, Double, String> l) {
DataSet<Tuple2<Integer, Double>> revenueByCustomer =
ordersFilteredByYear.joinWithHuge(lineitemsFilteredByFlag)
.where(0).equalTo(0)
- .projectFirst(1).projectSecond(1)
- .types(Integer.class, Double.class)
--- End diff --
This example still contains calls to `types(Class<?>...)`. Is this on purpose? I think we should have the call out of all examples. Having them in the tests is OK though. :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20136871
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/WebLogAnalysisWithoutType.java ---
@@ -0,0 +1,334 @@
+/*
+ * 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.flink.examples.java.relational;
+
+
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.util.Collector;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.examples.java.relational.util.WebLogData;
+import org.apache.flink.examples.java.relational.util.WebLogDataGenerator;
+
+/**
+ * This program processes web logs and relational data.
+ * It implements the following relational query:
+ *
+ * <code><pre>
+ * SELECT
+ * r.pageURL,
+ * r.pageRank,
+ * r.avgDuration
+ * FROM documents d JOIN rankings r
+ * ON d.url = r.url
+ * WHERE CONTAINS(d.text, [keywords])
+ * AND r.rank > [rank]
+ * AND NOT EXISTS
+ * (
+ * SELECT * FROM Visits v
+ * WHERE v.destUrl = d.url
+ * AND v.visitDate < [date]
+ * );
+ * </pre></code>
+ *
+ * <p>
+ * Input files are plain text CSV files using the pipe character ('|') as field separator.
+ * The tables referenced in the query can be generated using the {@link WebLogDataGenerator} and
+ * have the following schemas
+ * <code><pre>
+ * CREATE TABLE Documents (
+ * url VARCHAR(100) PRIMARY KEY,
+ * contents TEXT );
+ *
+ * CREATE TABLE Rankings (
+ * pageRank INT,
+ * pageURL VARCHAR(100) PRIMARY KEY,
+ * avgDuration INT );
+ *
+ * CREATE TABLE Visits (
+ * sourceIP VARCHAR(16),
+ * destURL VARCHAR(100),
+ * visitDate DATE,
+ * adRevenue FLOAT,
+ * userAgent VARCHAR(64),
+ * countryCode VARCHAR(3),
+ * languageCode VARCHAR(6),
+ * searchWord VARCHAR(32),
+ * duration INT );
+ * </pre></code>
+ *
+ * <p>
+ * Usage: <code>WebLogAnalysis <documents path> <ranks path> <visits path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WebLogData}.
+ *
+ * <p>
+ * This example shows how to use:
+ * <ul>
+ * <li> tuple data types
+ * <li> projection and join projection
+ * <li> the CoGroup transformation for an anti-join
+ * </ul>
+ *
+ */
+@SuppressWarnings("serial")
+public class WebLogAnalysisWithoutType {
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(String[] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // get input data
+ DataSet<Tuple2<String, String>> documents = getDocumentsDataSet(env);
+ DataSet<Tuple3<Integer, String, Integer>> ranks = getRanksDataSet(env);
+ DataSet<Tuple2<String, String>> visits = getVisitsDataSet(env);
+
+ // Retain documents with keywords
+ /*DataSet<Tuple1<String>> filterDocs = documents
+ .filter(new FilterDocByKeyWords())
+ .projection(0); */
--- End diff --
Why did you comment this block out?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20569485
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -517,859 +542,426 @@ public CrossProjection(DataSet<I1> ds1, DataSet<I2> ds2, int[] firstFieldIndexes
// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
/**
+ * Chooses a projectTupleX according to the length of {@link CrossProjection#fieldIndexes}
+ *
+ * @return The projected DataSet.
+ *
+ * @see Projection
+ */
+ @SuppressWarnings("unchecked")
+ public <OUT extends Tuple> ProjectCross<I1, I2, OUT> projectTupleX() {
+ ProjectCross<I1, I2, OUT> projectionCross = null;
+
+ switch (fieldIndexes.length) {
+ case 1: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple1(); break;
+ case 2: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple2(); break;
+ case 3: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple3(); break;
+ case 4: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple4(); break;
+ case 5: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple5(); break;
+ case 6: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple6(); break;
+ case 7: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple7(); break;
+ case 8: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple8(); break;
+ case 9: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple9(); break;
+ case 10: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple10(); break;
+ case 11: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple11(); break;
+ case 12: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple12(); break;
+ case 13: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple13(); break;
+ case 14: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple14(); break;
+ case 15: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple15(); break;
+ case 16: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple16(); break;
+ case 17: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple17(); break;
+ case 18: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple18(); break;
+ case 19: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple19(); break;
+ case 20: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple20(); break;
+ case 21: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple21(); break;
+ case 22: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple22(); break;
+ case 23: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple23(); break;
+ case 24: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple24(); break;
+ case 25: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple25(); break;
--- End diff --
Should we add a default case here, which throws an IllegalStateException (although we check the indexes previously)?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20569311
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/CrossOperator.java ---
@@ -517,859 +542,426 @@ public CrossProjection(DataSet<I1> ds1, DataSet<I2> ds2, int[] firstFieldIndexes
// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
/**
+ * Chooses a projectTupleX according to the length of {@link CrossProjection#fieldIndexes}
+ *
+ * @return The projected DataSet.
+ *
+ * @see Projection
+ */
+ @SuppressWarnings("unchecked")
+ public <OUT extends Tuple> ProjectCross<I1, I2, OUT> projectTupleX() {
+ ProjectCross<I1, I2, OUT> projectionCross = null;
+
+ switch (fieldIndexes.length) {
+ case 1: projectionCross = (ProjectCross<I1, I2, OUT>) projectTuple1(); break;
--- End diff --
Can you add a warning to the TupleGenerator classes that this method depends on the number of available tuples? It's a corner case, but it might happen that the number of available tuples and this method get out of sync.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by dukechain <gi...@git.apache.org>.
Github user dukechain commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20139659
--- Diff: flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/relational/WebLogAnalysisWithoutType.java ---
@@ -0,0 +1,334 @@
+/*
+ * 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.flink.examples.java.relational;
+
+
+import org.apache.flink.api.common.functions.CoGroupFunction;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.util.Collector;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.examples.java.relational.util.WebLogData;
+import org.apache.flink.examples.java.relational.util.WebLogDataGenerator;
+
+/**
+ * This program processes web logs and relational data.
+ * It implements the following relational query:
+ *
+ * <code><pre>
+ * SELECT
+ * r.pageURL,
+ * r.pageRank,
+ * r.avgDuration
+ * FROM documents d JOIN rankings r
+ * ON d.url = r.url
+ * WHERE CONTAINS(d.text, [keywords])
+ * AND r.rank > [rank]
+ * AND NOT EXISTS
+ * (
+ * SELECT * FROM Visits v
+ * WHERE v.destUrl = d.url
+ * AND v.visitDate < [date]
+ * );
+ * </pre></code>
+ *
+ * <p>
+ * Input files are plain text CSV files using the pipe character ('|') as field separator.
+ * The tables referenced in the query can be generated using the {@link WebLogDataGenerator} and
+ * have the following schemas
+ * <code><pre>
+ * CREATE TABLE Documents (
+ * url VARCHAR(100) PRIMARY KEY,
+ * contents TEXT );
+ *
+ * CREATE TABLE Rankings (
+ * pageRank INT,
+ * pageURL VARCHAR(100) PRIMARY KEY,
+ * avgDuration INT );
+ *
+ * CREATE TABLE Visits (
+ * sourceIP VARCHAR(16),
+ * destURL VARCHAR(100),
+ * visitDate DATE,
+ * adRevenue FLOAT,
+ * userAgent VARCHAR(64),
+ * countryCode VARCHAR(3),
+ * languageCode VARCHAR(6),
+ * searchWord VARCHAR(32),
+ * duration INT );
+ * </pre></code>
+ *
+ * <p>
+ * Usage: <code>WebLogAnalysis <documents path> <ranks path> <visits path> <result path></code><br>
+ * If no parameters are provided, the program is run with default data from {@link WebLogData}.
+ *
+ * <p>
+ * This example shows how to use:
+ * <ul>
+ * <li> tuple data types
+ * <li> projection and join projection
+ * <li> the CoGroup transformation for an anti-join
+ * </ul>
+ *
+ */
+@SuppressWarnings("serial")
+public class WebLogAnalysisWithoutType {
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(String[] args) throws Exception {
+
+ if(!parseParameters(args)) {
+ return;
+ }
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // get input data
+ DataSet<Tuple2<String, String>> documents = getDocumentsDataSet(env);
+ DataSet<Tuple3<Integer, String, Integer>> ranks = getRanksDataSet(env);
+ DataSet<Tuple2<String, String>> visits = getVisitsDataSet(env);
+
+ // Retain documents with keywords
+ /*DataSet<Tuple1<String>> filterDocs = documents
+ .filter(new FilterDocByKeyWords())
+ .projection(0); */
--- End diff --
Here, I just wanna check whether it makes sense if we do not specify "Tuple1<String>" for "filterDocs". However, there are errors when it meets "with" as user-defined functions usually need to know "TupleX" to process the dataset. For example, if we specify DataSet<Tuple> joinDocsRanks, it would be a error in "joinDocsRanks.coGroup(filterVisits)....with(new AntiJoinVisits())". I would like to revert it to the commented block.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-62519734
We can keep it source compatible by keeping the 'types()' call and deprecating it. The call before would produce a ProjectedDataSet that is already valid, and the deprecated types call could be added optionally.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-64593654
While trying around locally I noticed a problem. The current master catches the type mismatch in the following program during compilation time, whereas the new branch does not. The resulting output is still correct, but we want to still be able to catch these mismatches, no?
```java
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
Tuple2<Integer, Integer>[] data = new Tuple2Builder<Integer, Integer>()
.add(0, 1)
.add(2, 3)
.add(4, 5)
.build();
DataSet<Tuple2<Integer, Integer>> input = env.fromElements(data);
DataSet<Tuple2<Integer, String>> output = input.project(0, 0).types(Integer.class, Integer.class);
output.print();
env.execute();
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20142156
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -93,6 +113,30 @@ public Projection(DataSet<T> ds, int[] fieldIndexes) {
this.fieldIndexes = fieldIndexes;
}
+ public void acceptAdditionalIndexes(int... additionalIndexes) {
--- End diff --
I see, good point.
Can you make the method private, such that it is not visible from the API?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20149433
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/JoinOperator.java ---
@@ -1215,8 +1735,7 @@ public JoinProjection(DataSet<I1> ds1, DataSet<I2> ds2, Keys<I1> keys1, Keys<I2>
* Projects a pair of joined elements to a {@link Tuple} with the previously selected fields.
* Requires the classes of the fields of the resulting tuples.
*
- * @param type0 The class of field '0' of the result tuples.
- * @param type1 The class of field '1' of the result tuples.
+ * @param type0 The class of field '0' of the result tuples. @param type1 The class of field '1' of the result tuples.
--- End diff --
Yes, sure. The ``types()`` methods should go :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20137140
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/DataSet.java ---
@@ -246,6 +247,24 @@ public ExecutionEnvironment getExecutionEnvironment() {
return new Projection<T>(this, fieldIndexes);
}
+ /**
+ * Initiates a Project transformation on a {@link Tuple} {@link DataSet}.<br/>
+ * <b>Note: Only Tuple DataSets can be projected.</b></br>
+ * The transformation projects each Tuple of the DataSet onto a (sub)set of fields.</br>
+ * This method returns a {@link ProjectOperator} to complete the transformation.
+ *
+ * @param fieldIndexes The field indexes of the input tuples that are retained.
+ * The order of fields in the output tuple corresponds to the order of field indexes.
+ * @return A ProjectOperator to complete the Project transformation.
+ *
+ * @see Tuple
+ * @see DataSet
+ * @see org.apache.flink.api.java.operators.ProjectOperator
+ */
+ public <OUT extends Tuple> ProjectOperator<?, OUT> projection(int... fieldIndexes) {
--- End diff --
Having ``project()`` and ``projection()`` methods might confuse users. Since this is an evolution of the original ```project()`` method, I'd keep that name and simply remove the ``type()`` method in ``ProjectionOperator`` although this is an API breaking change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/incubator-flink/pull/194#issuecomment-63635220
This looks good :) Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20138246
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -93,6 +113,30 @@ public Projection(DataSet<T> ds, int[] fieldIndexes) {
this.fieldIndexes = fieldIndexes;
}
+ public void acceptAdditionalIndexes(int... additionalIndexes) {
--- End diff --
I would prefer to have a slim API.
Do you have a specific use case in mind for adding this method to the API? Why shouldn't all indexes be set in the initial ``project()`` call?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20138315
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java ---
@@ -323,9 +534,12 @@ private static void modifyJoinProjectOperator(File root) throws IOException {
sb.append("\t\t * Projects a pair of joined elements to a {@link Tuple} with the previously selected fields. \n");
sb.append("\t\t * Requires the classes of the fields of the resulting tuples. \n");
sb.append("\t\t * \n");
+ sb.append("\t\t *");
for (int i = 0; i < numFields; i++) {
- sb.append("\t\t * @param type" + i + " The class of field '"+i+"' of the result tuples.\n");
+ sb.append(" @param type" + i + " The class of field '"+i+"' of the result tuples. ");
--- End diff --
This looks like the change that removed the new line and reformatted the JavaDocs.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [FLINK-1040] Make type() call in pro...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/194#discussion_r20572664
--- Diff: flink-java/src/main/java/org/apache/flink/api/java/operators/ProjectOperator.java ---
@@ -63,11 +73,28 @@ public ProjectOperator(DataSet<IN> input, int[] fields, TupleTypeInfo<OUT> retur
return ppo;
}
-
+
+ @SuppressWarnings("hiding")
+ public <OUT extends Tuple> ProjectOperator<?, OUT> project(int... fieldIndexes) {
+ proj.acceptAdditionalIndexes(fieldIndexes);
+
+ return proj.projectTupleX();
+ }
+ /**
+ * A fake types() call to make codes compatible
+ * @param types
+ * @return
+ */
+ @SuppressWarnings({ "unchecked", "hiding" })
+ @Deprecated
+ public <OUT extends Tuple> ProjectOperator<?, OUT> types(Class<?>... types) {
+ return (ProjectOperator<?, OUT>) this;
+ }
+
public static class Projection<T> {
- private final DataSet<T> ds;
- private final int[] fieldIndexes;
+ private final DataSet<T> ds;
--- End diff --
trivial: tab
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---