There is no reason for it, the operator and function doesn't rely on the logic which AbstractUdfStreamOperator supplied.
Best, Kurt On Wed, Apr 17, 2019 at 4:35 PM Felipe Gutierrez < felipe.o.gutier...@gmail.com> wrote: > Thanks for the tip! I guess now it is working as it should be > <https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/operators/MapBundleFunctionImpl.java#L54> > . > > Just one last question. Why did you decide to use "AbstractStreamOperator" > instead of "AbstractUdfStreamOperator". I am asking because I was basing my > solution also (I also looked at your solution) on the "StreamFlatMap > <https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamFlatMap.java#L29>" > class implementation. > > Best, > Felipe > > *--* > *-- Felipe Gutierrez* > > *-- skype: felipe.o.gutierrez* > *--* *https://felipeogutierrez.blogspot.com > <https://felipeogutierrez.blogspot.com>* > > > On Wed, Apr 17, 2019 at 4:13 AM Kurt Young <ykt...@gmail.com> wrote: > >> I think you might mixed some test codes with the operator. "List<String> >> getOutputs()" is from "TestMapBundleFunction" and only used for >> validation. >> For the real usage, you need to write whatever records you want to emit >> to the "collector" which passed in during "finishBundle". >> >> Best, >> Kurt >> >> >> On Wed, Apr 17, 2019 at 12:50 AM Felipe Gutierrez < >> felipe.o.gutier...@gmail.com> wrote: >> >>> Hi Kurt, >>> >>> How do you make the finishBundle >>> <https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/bundle/MapBundleFunction.java#L59> >>> method returns the combined tuples? I saw that there is a method >>> "List<String> getOutputs()" which is never called. >>> >>> I did an implementation >>> <https://github.com/felipegutierrez/explore-flink/tree/master/src/main/java/org/sense/flink/examples/stream/operators> >>> based on the example that you suggested. The MapBundleFunctionImpl >>> <https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/operators/MapBundleFunctionImpl.java#L53> >>> class >>> has the method finishBundle which iterate all the combined tuples and >>> return it. However, my application does not continue to receive tuples >>> after the transform method >>> <https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/MqttSensorDataSkewedCombinerByKeySkewedDAG.java#L86> >>> . >>> >>> Thanks, >>> Felipe >>> >>> *--* >>> *-- Felipe Gutierrez* >>> >>> *-- skype: felipe.o.gutierrez* >>> *--* *https://felipeogutierrez.blogspot.com >>> <https://felipeogutierrez.blogspot.com>* >>> >>> >>> On Tue, Apr 16, 2019 at 3:10 AM Kurt Young <ykt...@gmail.com> wrote: >>> >>>> I think you can simply copy the source codes to your project if maven >>>> dependency can not be used. >>>> >>>> Best, >>>> Kurt >>>> >>>> >>>> On Mon, Apr 15, 2019 at 9:47 PM Felipe Gutierrez < >>>> felipe.o.gutier...@gmail.com> wrote: >>>> >>>>> Hi again Kurt, >>>>> >>>>> could you please help me with the pom.xml file? I have included >>>>> all Table ecosystem dependencies and the flink-table-runtime-blink as >>>>> well. >>>>> However the class org.apache.flink.table.runtime.context.ExecutionContext >>>>> is still not found. I guess I am missing some dependency, but I do not >>>>> know >>>>> which. This is my pom.xml file. >>>>> >>>>> <project xmlns="http://maven.apache.org/POM/4.0.0"; >>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; >>>>> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 >>>>> http://maven.apache.org/xsd/maven-4.0.0.xsd";> >>>>> <modelVersion>4.0.0</modelVersion> >>>>> >>>>> <groupId>org.sense.flink</groupId> >>>>> <artifactId>explore-flink</artifactId> >>>>> <version>0.0.1-SNAPSHOT</version> >>>>> <packaging>jar</packaging> >>>>> >>>>> <name>explore-flink</name> >>>>> <url>http://maven.apache.org</url> >>>>> >>>>> <properties> >>>>> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> >>>>> <jdk.version>1.8</jdk.version> >>>>> <scala.binary.version>2.11</scala.binary.version> >>>>> <!-- <flink.version>1.8.0</flink.version> --> >>>>> <flink.version>1.9-SNAPSHOT</flink.version> >>>>> <junit.version>4.12</junit.version> >>>>> </properties> >>>>> >>>>> <dependencies> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-java</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-clients_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-metrics-dropwizard</artifactId> >>>>> <version>${flink.version}</version> >>>>> <scope>provided</scope> >>>>> </dependency> >>>>> >>>>> <!-- Table ecosystem --> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> >>>>> <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> >>>>> <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-table-planner_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.apache.flink</groupId> >>>>> <artifactId>flink-table-runtime-blink</artifactId> >>>>> <version>${flink.version}</version> >>>>> </dependency> >>>>> >>>>> <dependency> >>>>> <groupId>org.fusesource.mqtt-client</groupId> >>>>> <artifactId>mqtt-client</artifactId> >>>>> <version>1.15</version> >>>>> <!-- <scope>provided</scope> --> >>>>> </dependency> >>>>> >>>>> <dependency> >>>>> <groupId>org.slf4j</groupId> >>>>> <artifactId>slf4j-api</artifactId> >>>>> <version>1.7.26</version> >>>>> </dependency> >>>>> <dependency> >>>>> <groupId>org.slf4j</groupId> >>>>> <artifactId>slf4j-log4j12</artifactId> >>>>> <version>1.7.26</version> >>>>> </dependency> >>>>> >>>>> <dependency> >>>>> <groupId>junit</groupId> >>>>> <artifactId>junit</artifactId> >>>>> <version>${junit.version}</version> >>>>> </dependency> >>>>> </dependencies> >>>>> <build> >>>>> <finalName>explore-flink</finalName> >>>>> <plugins> >>>>> <!-- download source code in Eclipse, best practice --> >>>>> <plugin> >>>>> <groupId>org.apache.maven.plugins</groupId> >>>>> <artifactId>maven-eclipse-plugin</artifactId> >>>>> <version>2.10</version> >>>>> <configuration> >>>>> <downloadSources>true</downloadSources> >>>>> <downloadJavadocs>false</downloadJavadocs> >>>>> </configuration> >>>>> </plugin> >>>>> >>>>> <!-- Set a compiler level --> >>>>> <plugin> >>>>> <groupId>org.apache.maven.plugins</groupId> >>>>> <artifactId>maven-compiler-plugin</artifactId> >>>>> <version>3.8.0</version> >>>>> <configuration> >>>>> <source>${jdk.version}</source> >>>>> <target>${jdk.version}</target> >>>>> </configuration> >>>>> </plugin> >>>>> >>>>> <!-- Maven Shade Plugin --> >>>>> <plugin> >>>>> <groupId>org.apache.maven.plugins</groupId> >>>>> <artifactId>maven-shade-plugin</artifactId> >>>>> <version>3.2.0</version> >>>>> <!-- Run shade goal on package phase --> >>>>> <executions> >>>>> <execution> >>>>> <phase>package</phase> >>>>> <goals> >>>>> <goal>shade</goal> >>>>> </goals> >>>>> <configuration> >>>>> <artifactSet> >>>>> <excludes> >>>>> <exclude>org.apache.flink:*</exclude> >>>>> <!-- Also exclude very big transitive dependencies of Flink WARNING: >>>>> You have to remove these excludes if your code relies on other >>>>> versions of >>>>> these dependencies. --> >>>>> <exclude>org.slf4j:*</exclude> >>>>> <exclude>log4j:*</exclude> >>>>> <exclude>com.typesafe:config:*</exclude> >>>>> <exclude>junit:junit:*</exclude> >>>>> <exclude>com.codahale.metrics:*</exclude> >>>>> </excludes> >>>>> </artifactSet> >>>>> <filters> >>>>> <filter> >>>>> <artifact>org.apache.flink:*</artifact> >>>>> <excludes> >>>>> <!-- exclude shaded google but include shaded curator --> >>>>> <exclude>org/apache/flink/shaded/com/**</exclude> >>>>> <exclude>web-docs/**</exclude> >>>>> </excludes> >>>>> </filter> >>>>> <filter> >>>>> <!-- Do not copy the signatures in the META-INF folder. Otherwise, >>>>> this might cause SecurityExceptions when using the JAR. --> >>>>> <artifact>*:*</artifact> >>>>> <excludes> >>>>> <exclude>META-INF/*.SF</exclude> >>>>> <exclude>META-INF/*.DSA</exclude> >>>>> <exclude>META-INF/*.RSA</exclude> >>>>> </excludes> >>>>> </filter> >>>>> <filter> >>>>> <artifact>*:*</artifact> >>>>> <includes> >>>>> <include>org/apache/calcite/**</include> >>>>> <include>org/apache/flink/calcite/shaded/**</include> >>>>> <include>org/apache/flink/table/**</include> >>>>> <include>org.codehaus.commons.compiler.properties</include> >>>>> <include>org/codehaus/janino/**</include> >>>>> <include>org/codehaus/commons/**</include> >>>>> </includes> >>>>> </filter> >>>>> </filters> >>>>> <!-- If you want to use ./bin/flink run <quickstart jar> uncomment >>>>> the following lines. This will add a Main-Class entry to the manifest >>>>> file --> >>>>> <transformers> >>>>> <transformer >>>>> >>>>> implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> >>>>> <mainClass>org.sense.flink.App</mainClass> >>>>> </transformer> >>>>> </transformers> >>>>> <createDependencyReducedPom>false</createDependencyReducedPom> >>>>> </configuration> >>>>> </execution> >>>>> </executions> >>>>> </plugin> >>>>> </plugins> >>>>> </build> >>>>> </project> >>>>> >>>>> >>>>> Thanks >>>>> >>>>> >>>>> >>>>> *--* >>>>> *-- Felipe Gutierrez* >>>>> >>>>> *-- skype: felipe.o.gutierrez* >>>>> *--* *https://felipeogutierrez.blogspot.com >>>>> <https://felipeogutierrez.blogspot.com>* >>>>> >>>>> >>>>> On Mon, Apr 15, 2019 at 3:25 PM Felipe Gutierrez < >>>>> felipe.o.gutier...@gmail.com> wrote: >>>>> >>>>>> oh, yes. I just saw. I will use 1.9 then. thanks >>>>>> >>>>>> *--* >>>>>> *-- Felipe Gutierrez* >>>>>> >>>>>> *-- skype: felipe.o.gutierrez* >>>>>> *--* *https://felipeogutierrez.blogspot.com >>>>>> <https://felipeogutierrez.blogspot.com>* >>>>>> >>>>>> >>>>>> On Mon, Apr 15, 2019 at 3:23 PM Kurt Young <ykt...@gmail.com> wrote: >>>>>> >>>>>>> It's because all blink codes are not shipped with 1.8.0, they >>>>>>> current only available in 1.9-SNAPSHOT. >>>>>>> >>>>>>> Best, >>>>>>> Kurt >>>>>>> >>>>>>> >>>>>>> On Mon, Apr 15, 2019 at 7:18 PM Felipe Gutierrez < >>>>>>> felipe.o.gutier...@gmail.com> wrote: >>>>>>> >>>>>>>> Hi, >>>>>>>> >>>>>>>> what are the artifacts that I have to import on maven in order to >>>>>>>> use Blink Api? >>>>>>>> >>>>>>>> I am using Flink 1.8.0 and I am trying to import blink code to use >>>>>>>> its ExecutionContext >>>>>>>> <https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/context/ExecutionContext.java>. >>>>>>>> I want to do this in order to implement my own operator like it is >>>>>>>> implemented here >>>>>>>> <https://github.com/apache/flink/blob/blink/flink-libraries/flink-table/src/main/java/org/apache/flink/table/runtime/bundle/BundleOperator.java>. >>>>>>>> I guess if I import flink-table everything should come inside the same >>>>>>>> jar >>>>>>>> as it is done here >>>>>>>> <https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/pom.xml>. >>>>>>>> However, I cannot import "flink-table-runtime-blink". Eclipse says >>>>>>>> that it >>>>>>>> is a missing artifact. >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-table-planner_2.11</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-table-api-java-bridge_2.11</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-streaming-scala_2.11</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-table-common</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-table</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> <type>pom</type> >>>>>>>> <scope>provided</scope> >>>>>>>> </dependency> >>>>>>>> <dependency> <!-- THIS IS NOT POSSIBLE TO IMPORT --> >>>>>>>> <groupId>org.apache.flink</groupId> >>>>>>>> <artifactId>flink-table-runtime-blink</artifactId> >>>>>>>> <version>1.8.0</version> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> >>>>>>>> *--* >>>>>>>> *-- Felipe Gutierrez* >>>>>>>> >>>>>>>> *-- skype: felipe.o.gutierrez* >>>>>>>> *--* *https://felipeogutierrez.blogspot.com >>>>>>>> <https://felipeogutierrez.blogspot.com>* >>>>>>>> >>>>>>>> >>>>>>>> On Mon, Apr 15, 2019 at 9:49 AM Felipe Gutierrez < >>>>>>>> felipe.o.gutier...@gmail.com> wrote: >>>>>>>> >>>>>>>>> Cool, thanks Kurt! >>>>>>>>> *-* >>>>>>>>> *- Felipe Gutierrez* >>>>>>>>> >>>>>>>>> *- skype: felipe.o.gutierrez* >>>>>>>>> *- **https://felipeogutierrez.blogspot.com >>>>>>>>> <https://felipeogutierrez.blogspot.com>* * >>>>>>>>> <https://felipeogutierrez.blogspot.com>* >>>>>>>>> >>>>>>>>> >>>>>>>>> On Mon, Apr 15, 2019 at 6:06 AM Kurt Young <ykt...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> Hi, >>>>>>>>>> >>>>>>>>>> You can checkout the bundle operator which used in Blink to >>>>>>>>>> perform similar thing you mentioned: >>>>>>>>>> https://github.com/apache/flink/blob/blink/flink-libraries/flink-table/src/main/java/org/apache/flink/table/runtime/bundle/BundleOperator.java >>>>>>>>>> >>>>>>>>>> Best, >>>>>>>>>> Kurt >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Fri, Apr 12, 2019 at 8:05 PM Felipe Gutierrez < >>>>>>>>>> felipe.o.gutier...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>>> Hi, >>>>>>>>>>> >>>>>>>>>>> I was trying to implement a better way to handle data skew using >>>>>>>>>>> Flink and I found this talk from #FlinkForward SF 2017: "Cliff >>>>>>>>>>> Resnick & Seth Wiesman - From Zero to Streaming >>>>>>>>>>> <https://youtu.be/mSLesPzWplA?t=835>" [1] which says that they >>>>>>>>>>> used OneInputStreamOperator [2]. Through it, they could implement >>>>>>>>>>> the >>>>>>>>>>> "combiner" in Hadoop (execute part of the reduce tasks on the Map >>>>>>>>>>> phase, >>>>>>>>>>> before shuffling). >>>>>>>>>>> >>>>>>>>>>> I need some help here. What are some of the Flink source-code >>>>>>>>>>> operators that I can peek up to implement my on operator that deals >>>>>>>>>>> with >>>>>>>>>>> data skew? Or maybe, is there someone that have an example of a use >>>>>>>>>>> case >>>>>>>>>>> similar to this? >>>>>>>>>>> >>>>>>>>>>> [1] https://youtu.be/mSLesPzWplA?t=835 >>>>>>>>>>> [2] >>>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.html >>>>>>>>>>> >>>>>>>>>>> Thanks! >>>>>>>>>>> Felipe >>>>>>>>>>> >>>>>>>>>>> *--* >>>>>>>>>>> *-- Felipe Gutierrez* >>>>>>>>>>> >>>>>>>>>>> *-- skype: felipe.o.gutierrez* >>>>>>>>>>> *--* *https://felipeogutierrez.blogspot.com >>>>>>>>>>> <https://felipeogutierrez.blogspot.com>* >>>>>>>>>>> >>>>>>>>>>
