You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/01/14 17:16:02 UTC
[05/50] flink git commit: [FLINK-3195] [examples] Consolidate batch
examples into one project,
unify batch and streaming examples under on parent project
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/util/WordCountData.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/util/WordCountData.java b/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/util/WordCountData.java
deleted file mode 100644
index 52efee7..0000000
--- a/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/util/WordCountData.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.examples.java.wordcount.util;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-
-/**
- * Provides the default data sets used for the WordCount example program.
- * The default data sets are used, if no parameters are given to the program.
- *
- */
-public class WordCountData {
-
- public static final String[] WORDS = new String[] {
- "To be, or not to be,--that is the question:--",
- "Whether 'tis nobler in the mind to suffer",
- "The slings and arrows of outrageous fortune",
- "Or to take arms against a sea of troubles,",
- "And by opposing end them?--To die,--to sleep,--",
- "No more; and by a sleep to say we end",
- "The heartache, and the thousand natural shocks",
- "That flesh is heir to,--'tis a consummation",
- "Devoutly to be wish'd. To die,--to sleep;--",
- "To sleep! perchance to dream:--ay, there's the rub;",
- "For in that sleep of death what dreams may come,",
- "When we have shuffled off this mortal coil,",
- "Must give us pause: there's the respect",
- "That makes calamity of so long life;",
- "For who would bear the whips and scorns of time,",
- "The oppressor's wrong, the proud man's contumely,",
- "The pangs of despis'd love, the law's delay,",
- "The insolence of office, and the spurns",
- "That patient merit of the unworthy takes,",
- "When he himself might his quietus make",
- "With a bare bodkin? who would these fardels bear,",
- "To grunt and sweat under a weary life,",
- "But that the dread of something after death,--",
- "The undiscover'd country, from whose bourn",
- "No traveller returns,--puzzles the will,",
- "And makes us rather bear those ills we have",
- "Than fly to others that we know not of?",
- "Thus conscience does make cowards of us all;",
- "And thus the native hue of resolution",
- "Is sicklied o'er with the pale cast of thought;",
- "And enterprises of great pith and moment,",
- "With this regard, their currents turn awry,",
- "And lose the name of action.--Soft you now!",
- "The fair Ophelia!--Nymph, in thy orisons",
- "Be all my sins remember'd."
- };
-
- public static DataSet<String> getDefaultTextLineDataSet(ExecutionEnvironment env) {
- return env.fromElements(WORDS);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-java-examples/src/main/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/resources/log4j-test.properties b/flink-examples/flink-java-examples/src/main/resources/log4j-test.properties
deleted file mode 100644
index 65bd0b8..0000000
--- a/flink-examples/flink-java-examples/src/main/resources/log4j-test.properties
+++ /dev/null
@@ -1,23 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-log4j.rootLogger=INFO, console
-
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-java-examples/src/main/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/resources/log4j.properties b/flink-examples/flink-java-examples/src/main/resources/log4j.properties
deleted file mode 100644
index da32ea0..0000000
--- a/flink-examples/flink-java-examples/src/main/resources/log4j.properties
+++ /dev/null
@@ -1,23 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-log4j.rootLogger=INFO, console
-
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-java-examples/src/main/resources/logback.xml
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/resources/logback.xml b/flink-examples/flink-java-examples/src/main/resources/logback.xml
deleted file mode 100644
index 95f2d04..0000000
--- a/flink-examples/flink-java-examples/src/main/resources/logback.xml
+++ /dev/null
@@ -1,29 +0,0 @@
-<!--
- ~ Licensed to the Apache Software Foundation (ASF) under one
- ~ or more contributor license agreements. See the NOTICE file
- ~ distributed with this work for additional information
- ~ regarding copyright ownership. The ASF licenses this file
- ~ to you under the Apache License, Version 2.0 (the
- ~ "License"); you may not use this file except in compliance
- ~ with the License. You may obtain a copy of the License at
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing, software
- ~ distributed under the License is distributed on an "AS IS" BASIS,
- ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- ~ See the License for the specific language governing permissions and
- ~ limitations under the License.
- -->
-
-<configuration>
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern>
- </encoder>
- </appender>
-
- <root level="INFO">
- <appender-ref ref="STDOUT"/>
- </root>
-</configuration>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/pom.xml
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/pom.xml b/flink-examples/flink-scala-examples/pom.xml
deleted file mode 100644
index 7cb5f29..0000000
--- a/flink-examples/flink-scala-examples/pom.xml
+++ /dev/null
@@ -1,487 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<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/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-examples</artifactId>
- <version>1.0-SNAPSHOT</version>
- <relativePath>..</relativePath>
- </parent>
-
- <artifactId>flink-scala-examples</artifactId>
- <name>flink-scala-examples</name>
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-scala</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java-examples</artifactId>
- <version>${project.version}</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <!-- Scala Compiler -->
- <plugin>
- <groupId>net.alchim31.maven</groupId>
- <artifactId>scala-maven-plugin</artifactId>
- <version>3.1.4</version>
- <executions>
- <!-- Run scala compiler in the process-resources phase, so that dependencies on
- scala classes can be resolved later in the (Java) compile phase -->
- <execution>
- <id>scala-compile-first</id>
- <phase>process-resources</phase>
- <goals>
- <goal>compile</goal>
- </goals>
- </execution>
-
- <!-- Run scala compiler in the process-test-resources phase, so that dependencies on
- scala classes can be resolved later in the (Java) test-compile phase -->
- <execution>
- <id>scala-test-compile</id>
- <phase>process-test-resources</phase>
- <goals>
- <goal>testCompile</goal>
- </goals>
- </execution>
- </executions>
- <configuration>
- <jvmArgs>
- <jvmArg>-Xms128m</jvmArg>
- <jvmArg>-Xmx512m</jvmArg>
- </jvmArgs>
- </configuration>
- </plugin>
-
- <!-- Eclipse Integration -->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-eclipse-plugin</artifactId>
- <version>2.8</version>
- <configuration>
- <downloadSources>true</downloadSources>
- <projectnatures>
- <projectnature>org.scala-ide.sdt.core.scalanature</projectnature>
- <projectnature>org.eclipse.jdt.core.javanature</projectnature>
- </projectnatures>
- <buildcommands>
- <buildcommand>org.scala-ide.sdt.core.scalabuilder</buildcommand>
- </buildcommands>
- <classpathContainers>
- <classpathContainer>org.scala-ide.sdt.launching.SCALA_CONTAINER</classpathContainer>
- <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
- </classpathContainers>
- <excludes>
- <exclude>org.scala-lang:scala-library</exclude>
- <exclude>org.scala-lang:scala-compiler</exclude>
- </excludes>
- <sourceIncludes>
- <sourceInclude>**/*.scala</sourceInclude>
- <sourceInclude>**/*.java</sourceInclude>
- </sourceIncludes>
- </configuration>
- </plugin>
-
- <!-- Adding scala source directories to build path -->
- <plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <version>1.7</version>
- <executions>
- <!-- Add src/main/scala to eclipse build path -->
- <execution>
- <id>add-source</id>
- <phase>generate-sources</phase>
- <goals>
- <goal>add-source</goal>
- </goals>
- <configuration>
- <sources>
- <source>src/main/scala</source>
- </sources>
- </configuration>
- </execution>
- <!-- Add src/test/scala to eclipse build path -->
- <execution>
- <id>add-test-source</id>
- <phase>generate-test-sources</phase>
- <goals>
- <goal>add-test-source</goal>
- </goals>
- <configuration>
- <sources>
- <source>src/test/scala</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
-
- <plugin>
- <groupId>org.scalastyle</groupId>
- <artifactId>scalastyle-maven-plugin</artifactId>
- <version>0.5.0</version>
- <executions>
- <execution>
- <goals>
- <goal>check</goal>
- </goals>
- </execution>
- </executions>
- <configuration>
- <verbose>false</verbose>
- <failOnViolation>true</failOnViolation>
- <includeTestSourceDirectory>true</includeTestSourceDirectory>
- <failOnWarning>false</failOnWarning>
- <sourceDirectory>${basedir}/src/main/scala</sourceDirectory>
- <testSourceDirectory>${basedir}/src/test/scala</testSourceDirectory>
- <configLocation>${project.basedir}/../../tools/maven/scalastyle-config.xml</configLocation>
- <outputFile>${project.basedir}/scalastyle-output.xml</outputFile>
- <outputEncoding>UTF-8</outputEncoding>
- </configuration>
- </plugin>
-
- <!-- get default data from flink-java-examples package -->
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-dependency-plugin</artifactId>
- <version>2.9</version><!--$NO-MVN-MAN-VER$-->
- <executions>
- <execution>
- <id>unpack</id>
- <phase>prepare-package</phase>
- <goals>
- <goal>unpack</goal>
- </goals>
- <configuration>
- <artifactItems>
- <artifactItem>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-java-examples</artifactId>
- <version>${project.version}</version>
- <type>jar</type>
- <overWrite>false</overWrite>
- <outputDirectory>${project.build.directory}/classes</outputDirectory>
- <includes>**/util/*Data*.class</includes>
- </artifactItem>
- </artifactItems>
- </configuration>
- </execution>
- </executions>
- </plugin>
-
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <executions>
-
- <!-- KMeans -->
- <execution>
- <id>KMeans</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
-
- <configuration>
- <classifier>KMeans</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.clustering.KMeans</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/clustering/KMeans.class</include>
- <include>**/scala/clustering/KMeans$*.class</include>
- <include>**/java/clustering/util/KMeansDataGenerator.class</include>
- <include>**/java/clustering/util/KMeansData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- Transitive Closure -->
- <execution>
- <id>TransitiveClosure</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>TransitiveClosure</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.graph.TransitiveClosureNaive</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/graph/TransitiveClosureNaive.class</include>
- <include>**/scala/graph/TransitiveClosureNaive$*.class</include>
- <include>**/java/graph/util/ConnectedComponentsData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- Connected Components -->
- <execution>
- <id>ConnectedComponents</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>ConnectedComponents</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.graph.ConnectedComponents</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/graph/ConnectedComponents.class</include>
- <include>**/scala/graph/ConnectedComponents$*.class</include>
- <include>**/java/graph/util/ConnectedComponentsData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- EnumTriangles Basic -->
- <execution>
- <id>EnumTrianglesBasic</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>EnumTrianglesBasic</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.graph.EnumTrianglesBasic</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/graph/EnumTrianglesBasic.class</include>
- <include>**/scala/graph/EnumTrianglesBasic$*.class</include>
- <include>**/java/graph/util/EnumTrianglesData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- EnumTriangles Opt -->
- <execution>
- <id>EnumTrianglesOpt</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>EnumTrianglesOpt</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.graph.EnumTrianglesOpt</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/graph/EnumTrianglesOpt.class</include>
- <include>**/scala/graph/EnumTrianglesOpt$*.class</include>
- <include>**/java/graph/util/EnumTrianglesData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- PageRank Basic-->
- <execution>
- <id>PageRankBasic</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>PageRankBasic</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.graph.PageRankBasic</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/graph/PageRankBasic.class</include>
- <include>**/scala/graph/PageRankBasic$*.class</include>
- <include>**/java/graph/util/PageRankData.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- These queries are currently not self-contained -->
-
- <!-- TPC-H Query 10 -->
-
- <!--
- <execution>
- <id>TPCHQuery10</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>TPCHQuery10</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.relational.TPCHQuery10</program-class>
- </manifestEntries>
- </archive>
- <includes>
- <include>**/scala/relational/TPCHQuery10.class</include>
- <include>**/scala/relational/TPCHQuery10$*.class</include>
- </includes>
- </configuration>
- </execution> -->
-
- <!-- TPC-H Query 3 -->
- <!--
- <execution>
- <id>TPCHQuery3</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>TPCHQuery3</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.relational.TPCHQuery3</program-class>
- </manifestEntries>
- </archive>
- <includes>
- <include>**/scala/relational/TPCHQuery3.class</include>
- <include>**/scala/relational/TPCHQuery3$*.class</include>
- </includes>
- </configuration>
- </execution> -->
-
- <!-- WebLogAnalysis -->
- <execution>
- <id>WebLogAnalysis</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>WebLogAnalysis</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.relational.WebLogAnalysis</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/relational/WebLogAnalysis.class</include>
- <include>**/scala/relational/WebLogAnalysis$*.class</include>
- <include>**/java/relational/util/WebLogData.class</include>
- <include>**/java/relational/util/WebLogDataGenerator.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- WordCount -->
- <execution>
- <id>WordCount</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>WordCount</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.examples.scala.wordcount.WordCount</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>**/scala/wordcount/WordCount.class</include>
- <include>**/scala/wordcount/WordCount$*.class</include>
- <include>**/java/wordcount/util/WordCountData.class</include>
- </includes>
- </configuration>
- </execution>
-
- </executions>
- </plugin>
- </plugins>
-
- <pluginManagement>
- <plugins>
- <!-- This plugin's configuration is used to store Eclipse m2e settings only.
- It has no influence on the Maven build itself and simply suppresses errors in Eclipse.-->
- <plugin>
- <groupId>org.eclipse.m2e</groupId>
- <artifactId>lifecycle-mapping</artifactId>
- <version>1.0.0</version>
- <configuration>
- <lifecycleMappingMetadata>
- <pluginExecutions>
- <pluginExecution>
- <pluginExecutionFilter>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-dependency-plugin</artifactId>
- <versionRange>[2.9,)</versionRange>
- <goals>
- <goal>unpack</goal>
- </goals>
- </pluginExecutionFilter>
- <action>
- <ignore/>
- </action>
- </pluginExecution>
- </pluginExecutions>
- </lifecycleMappingMetadata>
- </configuration>
- </plugin>
- </plugins>
- </pluginManagement>
- </build>
-
-</project>
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/java/org/apache/flink/examples/scala/Dummy.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/java/org/apache/flink/examples/scala/Dummy.java b/flink-examples/flink-scala-examples/src/main/java/org/apache/flink/examples/scala/Dummy.java
deleted file mode 100644
index 1887a35..0000000
--- a/flink-examples/flink-scala-examples/src/main/java/org/apache/flink/examples/scala/Dummy.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.examples.scala;
-
-/**
- * This dummy class exists only to create an empty
- * javadoc.jar in the flink-scala-examples project.
- * This is required for passing the maven central sync requirements.
- */
-public class Dummy {}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/clustering/KMeans.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/clustering/KMeans.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/clustering/KMeans.scala
deleted file mode 100644
index 08a3e62..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/clustering/KMeans.scala
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.clustering
-
-import org.apache.flink.api.common.functions._
-import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields
-import org.apache.flink.api.scala._
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.examples.java.clustering.util.KMeansData
-
-import scala.collection.JavaConverters._
-
-/**
- * This example implements a basic K-Means clustering algorithm.
- *
- * K-Means is an iterative clustering algorithm and works as follows:
- * K-Means is given a set of data points to be clustered and an initial set of ''K'' cluster
- * centers.
- * In each iteration, the algorithm computes the distance of each data point to each cluster center.
- * Each point is assigned to the cluster center which is closest to it.
- * Subsequently, each cluster center is moved to the center (''mean'') of all points that have
- * been assigned to it.
- * The moved cluster centers are fed into the next iteration.
- * The algorithm terminates after a fixed number of iterations (as in this implementation)
- * or if cluster centers do not (significantly) move in an iteration.
- * This is the Wikipedia entry for the [[http://en.wikipedia
- * .org/wiki/K-means_clustering K-Means Clustering algorithm]].
- *
- * This implementation works on two-dimensional data points.
- * It computes an assignment of data points to cluster centers, i.e.,
- * each data point is annotated with the id of the final cluster (center) it belongs to.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Data points are represented as two double values separated by a blank character.
- * Data points are separated by newline characters.
- * For example `"1.2 2.3\n5.3 7.2\n"` gives two data points (x=1.2, y=2.3) and (x=5.3,
- * y=7.2).
- * - Cluster centers are represented by an integer id and a point value.
- * For example `"1 6.2 3.2\n2 2.9 5.7\n"` gives two centers (id=1, x=6.2,
- * y=3.2) and (id=2, x=2.9, y=5.7).
- *
- * Usage:
- * {{{
- * KMeans <points path> <centers path> <result path> <num iterations>
- * }}}
- * If no parameters are provided, the program is run with default data from
- * [[org.apache.flink.examples.java.clustering.util.KMeansData]]
- * and 10 iterations.
- *
- * This example shows how to use:
- *
- * - Bulk iterations
- * - Broadcast variables in bulk iterations
- * - Custom Java objects (PoJos)
- */
-object KMeans {
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val points: DataSet[Point] = getPointDataSet(env)
- val centroids: DataSet[Centroid] = getCentroidDataSet(env)
-
- val finalCentroids = centroids.iterate(numIterations) { currentCentroids =>
- val newCentroids = points
- .map(new SelectNearestCenter).withBroadcastSet(currentCentroids, "centroids")
- .map { x => (x._1, x._2, 1L) }.withForwardedFields("_1; _2")
- .groupBy(0)
- .reduce { (p1, p2) => (p1._1, p1._2.add(p2._2), p1._3 + p2._3) }.withForwardedFields("_1")
- .map { x => new Centroid(x._1, x._2.div(x._3)) }.withForwardedFields("_1->id")
- newCentroids
- }
-
- val clusteredPoints: DataSet[(Int, Point)] =
- points.map(new SelectNearestCenter).withBroadcastSet(finalCentroids, "centroids")
-
- if (fileOutput) {
- clusteredPoints.writeAsCsv(outputPath, "\n", " ")
- env.execute("Scala KMeans Example")
- }
- else {
- clusteredPoints.print()
- }
-
- }
-
- private def parseParameters(programArguments: Array[String]): Boolean = {
- if (programArguments.length > 0) {
- fileOutput = true
- if (programArguments.length == 4) {
- pointsPath = programArguments(0)
- centersPath = programArguments(1)
- outputPath = programArguments(2)
- numIterations = Integer.parseInt(programArguments(3))
-
- true
- }
- else {
- System.err.println("Usage: KMeans <points path> <centers path> <result path> <num " +
- "iterations>")
-
- false
- }
- }
- else {
- System.out.println("Executing K-Means example with default parameters and built-in default " +
- "data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" We provide a data generator to create synthetic input files for this " +
- "program.")
- System.out.println(" Usage: KMeans <points path> <centers path> <result path> <num " +
- "iterations>")
-
- true
- }
- }
-
- private def getPointDataSet(env: ExecutionEnvironment): DataSet[Point] = {
- if (fileOutput) {
- env.readCsvFile[(Double, Double)](
- pointsPath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- .map { x => new Point(x._1, x._2)}
- }
- else {
- val points = KMeansData.POINTS map {
- case Array(x, y) => new Point(x.asInstanceOf[Double], y.asInstanceOf[Double])
- }
- env.fromCollection(points)
- }
- }
-
- private def getCentroidDataSet(env: ExecutionEnvironment): DataSet[Centroid] = {
- if (fileOutput) {
- env.readCsvFile[(Int, Double, Double)](
- centersPath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1, 2))
- .map { x => new Centroid(x._1, x._2, x._3)}
- }
- else {
- val centroids = KMeansData.CENTROIDS map {
- case Array(id, x, y) =>
- new Centroid(id.asInstanceOf[Int], x.asInstanceOf[Double], y.asInstanceOf[Double])
- }
- env.fromCollection(centroids)
- }
- }
-
- private var fileOutput: Boolean = false
- private var pointsPath: String = null
- private var centersPath: String = null
- private var outputPath: String = null
- private var numIterations: Int = 10
-
- /**
- * A simple two-dimensional point.
- */
- class Point(var x: Double, var y: Double) extends Serializable {
- def this() {
- this(0, 0)
- }
-
- def add(other: Point): Point = {
- x += other.x
- y += other.y
- this
- }
-
- def div(other: Long): Point = {
- x /= other
- y /= other
- this
- }
-
- def euclideanDistance(other: Point): Double = {
- Math.sqrt((x - other.x) * (x - other.x) + (y - other.y) * (y - other.y))
- }
-
- def clear(): Unit = {
- x = 0
- y = 0
- }
-
- override def toString: String = {
- x + " " + y
- }
- }
-
- /**
- * A simple two-dimensional centroid, basically a point with an ID.
- */
- class Centroid(var id: Int, x: Double, y: Double) extends Point(x, y) {
- def this() {
- this(0, 0, 0)
- }
-
- def this(id: Int, p: Point) {
- this(id, p.x, p.y)
- }
-
- override def toString: String = {
- id + " " + super.toString
- }
- }
-
- /** Determines the closest cluster center for a data point. */
- @ForwardedFields(Array("*->_2"))
- final class SelectNearestCenter extends RichMapFunction[Point, (Int, Point)] {
- private var centroids: Traversable[Centroid] = null
-
- /** Reads the centroid values from a broadcast variable into a collection. */
- override def open(parameters: Configuration) {
- centroids = getRuntimeContext.getBroadcastVariable[Centroid]("centroids").asScala
- }
-
- def map(p: Point): (Int, Point) = {
- var minDistance: Double = Double.MaxValue
- var closestCentroidId: Int = -1
- for (centroid <- centroids) {
- val distance = p.euclideanDistance(centroid)
- if (distance < minDistance) {
- minDistance = distance
- closestCentroidId = centroid.id
- }
- }
- (closestCentroidId, p)
- }
-
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala
deleted file mode 100644
index 38c7f40..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import org.apache.flink.api.scala._
-import org.apache.flink.examples.java.graph.util.ConnectedComponentsData
-import org.apache.flink.util.Collector
-
-/**
- * An implementation of the connected components algorithm, using a delta iteration.
- *
- * Initially, the algorithm assigns each vertex an unique ID. In each step, a vertex picks the
- * minimum of its own ID and its neighbors' IDs, as its new ID and tells its neighbors about its
- * new ID. After the algorithm has completed, all vertices in the same component will have the same
- * ID.
- *
- * A vertex whose component ID did not change needs not propagate its information in the next
- * step. Because of that, the algorithm is easily expressible via a delta iteration. We here model
- * the solution set as the vertices with their current component ids, and the workset as the changed
- * vertices. Because we see all vertices initially as changed, the initial workset and the initial
- * solution set are identical. Also, the delta to the solution set is consequently also the next
- * workset.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Vertices represented as IDs and separated by new-line characters. For example,
- * `"1\n2\n12\n42\n63"` gives five vertices (1), (2), (12), (42), and (63).
- * - Edges are represented as pairs for vertex IDs which are separated by space characters. Edges
- * are separated by new-line characters. For example `"1 2\n2 12\n1 12\n42 63"`
- * gives four (undirected) edges (1)-(2), (2)-(12), (1)-(12), and (42)-(63).
- *
- * Usage:
- * {{{
- * ConnectedComponents <vertices path> <edges path> <result path> <max number of iterations>
- * }}}
- *
- * If no parameters are provided, the program is run with default data from
- * [[org.apache.flink.examples.java.graph.util.ConnectedComponentsData]] and 10 iterations.
- *
- *
- * This example shows how to use:
- *
- * - Delta Iterations
- * - Generic-typed Functions
- *
- */
-object ConnectedComponents {
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
- // set up execution environment
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // read vertex and edge data
- // assign the initial components (equal to the vertex id)
- val vertices = getVerticesDataSet(env).map { id => (id, id) }.withForwardedFields("*->_1;*->_2")
-
- // undirected edges by emitting for each input edge the input edges itself and an inverted
- // version
- val edges = getEdgesDataSet(env).flatMap { edge => Seq(edge, (edge._2, edge._1)) }
-
- // open a delta iteration
- val verticesWithComponents = vertices.iterateDelta(vertices, maxIterations, Array("_1")) {
- (s, ws) =>
-
- // apply the step logic: join with the edges
- val allNeighbors = ws.join(edges).where(0).equalTo(0) { (vertex, edge) =>
- (edge._2, vertex._2)
- }.withForwardedFieldsFirst("_2->_2").withForwardedFieldsSecond("_2->_1")
-
- // select the minimum neighbor
- val minNeighbors = allNeighbors.groupBy(0).min(1)
-
- // update if the component of the candidate is smaller
- val updatedComponents = minNeighbors.join(s).where(0).equalTo(0) {
- (newVertex, oldVertex, out: Collector[(Long, Long)]) =>
- if (newVertex._2 < oldVertex._2) out.collect(newVertex)
- }.withForwardedFieldsFirst("*")
-
- // delta and new workset are identical
- (updatedComponents, updatedComponents)
- }
- if (fileOutput) {
- verticesWithComponents.writeAsCsv(outputPath, "\n", " ")
- env.execute("Scala Connected Components Example")
- } else {
- verticesWithComponents.print()
- }
-
- }
-
- private def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- fileOutput = true
- if (args.length == 4) {
- verticesPath = args(0)
- edgesPath = args(1)
- outputPath = args(2)
- maxIterations = args(3).toInt
-
- true
- } else {
- System.err.println("Usage: ConnectedComponents <vertices path> <edges path> <result path>" +
- " <max number of iterations>")
-
- false
- }
- } else {
- System.out.println("Executing Connected Components example with built-in default data.")
- System.out.println(" Provide parameters to read input data from a file.")
- System.out.println(" Usage: ConnectedComponents <vertices path> <edges path> <result path>" +
- " <max number of iterations>")
-
- true
- }
- }
-
- private def getVerticesDataSet(env: ExecutionEnvironment): DataSet[Long] = {
- if (fileOutput) {
- env.readCsvFile[Tuple1[Long]](
- verticesPath,
- includedFields = Array(0))
- .map { x => x._1 }
- }
- else {
- env.fromCollection(ConnectedComponentsData.VERTICES)
- }
- }
-
- private def getEdgesDataSet(env: ExecutionEnvironment): DataSet[(Long, Long)] = {
- if (fileOutput) {
- env.readCsvFile[(Long, Long)](
- edgesPath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- .map { x => (x._1, x._2)}
- }
- else {
- val edgeData = ConnectedComponentsData.EDGES map {
- case Array(x, y) => (x.asInstanceOf[Long], y.asInstanceOf[Long])
- }
- env.fromCollection(edgeData)
- }
- }
-
- private var fileOutput: Boolean = false
- private var verticesPath: String = null
- private var edgesPath: String = null
- private var maxIterations: Int = 10
- private var outputPath: String = null
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/DeltaPageRank.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/DeltaPageRank.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/DeltaPageRank.scala
deleted file mode 100644
index ae8a982..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/DeltaPageRank.scala
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import org.apache.flink.api.scala._
-import org.apache.flink.util.Collector
-import org.apache.flink.core.fs.FileSystem.WriteMode
-
-object DeltaPageRank {
-
- private final val DAMPENING_FACTOR: Double = 0.85
- private final val NUM_VERTICES = 5
- private final val INITIAL_RANK = 1.0 / NUM_VERTICES
- private final val RANDOM_JUMP = (1 - DAMPENING_FACTOR) / NUM_VERTICES
- private final val THRESHOLD = 0.0001 / NUM_VERTICES
-
- type Page = (Long, Double)
- type Adjacency = (Long, Array[Long])
-
- def main(args: Array[String]) {
-
- val maxIterations = 100
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val rawLines: DataSet[String] = env.fromElements(
- "1 2 3 4",
- "2 1",
- "3 5",
- "4 2 3",
- "5 2 4")
- val adjacency: DataSet[Adjacency] = rawLines
- .map(str => {
- val elements = str.split(' ')
- val id = elements(0).toLong
- val neighbors = elements.slice(1, elements.length).map(_.toLong)
- (id, neighbors)
- })
-
- val initialRanks: DataSet[Page] = adjacency.flatMap {
- (adj, out: Collector[Page]) =>
- {
- val targets = adj._2
- val rankPerTarget = INITIAL_RANK * DAMPENING_FACTOR / targets.length
-
- // dampend fraction to targets
- for (target <- targets) {
- out.collect((target, rankPerTarget))
- }
-
- // random jump to self
- out.collect((adj._1, RANDOM_JUMP))
- }
- }
- .groupBy(0).sum(1)
-
- val initialDeltas = initialRanks.map { (page) => (page._1, page._2 - INITIAL_RANK) }
- .withForwardedFields("_1")
-
- val iteration = initialRanks.iterateDelta(initialDeltas, maxIterations, Array(0)) {
-
- (solutionSet, workset) =>
- {
- val deltas = workset.join(adjacency).where(0).equalTo(0) {
- (lastDeltas, adj, out: Collector[Page]) =>
- {
- val targets = adj._2
- val deltaPerTarget = DAMPENING_FACTOR * lastDeltas._2 / targets.length
-
- for (target <- targets) {
- out.collect((target, deltaPerTarget))
- }
- }
- }
- .groupBy(0).sum(1)
- .filter(x => Math.abs(x._2) > THRESHOLD)
-
- val rankUpdates = solutionSet.join(deltas).where(0).equalTo(0) {
- (current, delta) => (current._1, current._2 + delta._2)
- }.withForwardedFieldsFirst("_1")
-
- (rankUpdates, deltas)
- }
- }
-
- iteration.print()
-
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesBasic.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesBasic.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesBasic.scala
deleted file mode 100644
index 4c05fbb..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesBasic.scala
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields
-import org.apache.flink.api.scala._
-import scala.collection.JavaConverters._
-import org.apache.flink.api.scala.ExecutionEnvironment
-import org.apache.flink.api.common.functions.GroupReduceFunction
-import org.apache.flink.util.Collector
-import org.apache.flink.examples.java.graph.util.EnumTrianglesData
-import org.apache.flink.api.common.operators.Order
-
-import scala.collection.mutable
-
-
-/**
- * Triangle enumeration is a pre-processing step to find closely connected parts in graphs.
- * A triangle consists of three edges that connect three vertices with each other.
- *
- * The algorithm works as follows:
- * It groups all edges that share a common vertex and builds triads, i.e., triples of vertices
- * that are connected by two edges. Finally, all triads are filtered for which no third edge exists
- * that closes the triangle.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Edges are represented as pairs for vertex IDs which are separated by space
- * characters. Edges are separated by new-line characters.
- * For example `"1 2\n2 12\n1 12\n42 63"` gives four (undirected) edges (1)-(2), (2)-(12),
- * (1)-(12), and (42)-(63) that include a triangle
- *
- * <pre>
- * (1)
- * / \
- * (2)-(12)
- * </pre>
- *
- * Usage:
- * {{{
- * EnumTriangleBasic <edge path> <result path>
- * }}}
- * <br>
- * If no parameters are provided, the program is run with default data from
- * [[org.apache.flink.examples.java.graph.util.EnumTrianglesData]]
- *
- * This example shows how to use:
- *
- * - Custom Java objects which extend Tuple
- * - Group Sorting
- *
- */
-object EnumTrianglesBasic {
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- // set up execution environment
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // read input data
- val edges = getEdgeDataSet(env)
-
- // project edges by vertex id
- val edgesById = edges map(e => if (e.v1 < e.v2) e else Edge(e.v2, e.v1) )
-
- val triangles = edgesById
- // build triads
- .groupBy("v1").sortGroup("v2", Order.ASCENDING).reduceGroup(new TriadBuilder())
- // filter triads
- .join(edgesById).where("v2", "v3").equalTo("v1", "v2") { (t, _) => t }
- .withForwardedFieldsFirst("*")
-
- // emit result
- if (fileOutput) {
- triangles.writeAsCsv(outputPath, "\n", ",")
- // execute program
- env.execute("TriangleEnumeration Example")
- } else {
- triangles.print()
- }
-
-
- }
-
- // *************************************************************************
- // USER DATA TYPES
- // *************************************************************************
-
- case class Edge(v1: Int, v2: Int) extends Serializable
- case class Triad(v1: Int, v2: Int, v3: Int) extends Serializable
-
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Builds triads (triples of vertices) from pairs of edges that share a vertex. The first vertex
- * of a triad is the shared vertex, the second and third vertex are ordered by vertexId. Assumes
- * that input edges share the first vertex and are in ascending order of the second vertex.
- */
- @ForwardedFields(Array("v1->v1"))
- class TriadBuilder extends GroupReduceFunction[Edge, Triad] {
-
- val vertices = mutable.MutableList[Integer]()
-
- override def reduce(edges: java.lang.Iterable[Edge], out: Collector[Triad]) = {
-
- // clear vertex list
- vertices.clear()
-
- // build and emit triads
- for(e <- edges.asScala) {
-
- // combine vertex with all previously read vertices
- for(v <- vertices) {
- out.collect(Triad(e.v1, v, e.v2))
- }
- vertices += e.v2
- }
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- fileOutput = true
- if (args.length == 2) {
- edgePath = args(0)
- outputPath = args(1)
-
- true
- } else {
- System.err.println("Usage: EnumTriangleBasic <edge path> <result path>")
-
- false
- }
- } else {
- System.out.println("Executing Enum Triangles Basic example with built-in default data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" Usage: EnumTriangleBasic <edge path> <result path>")
-
- true
- }
- }
-
- private def getEdgeDataSet(env: ExecutionEnvironment): DataSet[Edge] = {
- if (fileOutput) {
- env.readCsvFile[Edge](edgePath, fieldDelimiter = " ", includedFields = Array(0, 1))
- } else {
- val edges = EnumTrianglesData.EDGES.map {
- case Array(v1, v2) => new Edge(v1.asInstanceOf[Int], v2.asInstanceOf[Int])
- }
- env.fromCollection(edges)
- }
- }
-
-
- private var fileOutput: Boolean = false
- private var edgePath: String = null
- private var outputPath: String = null
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesOpt.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesOpt.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesOpt.scala
deleted file mode 100644
index ad7e3a4..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/EnumTrianglesOpt.scala
+++ /dev/null
@@ -1,253 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields
-import org.apache.flink.api.scala._
-import scala.collection.JavaConverters._
-import org.apache.flink.api.scala.ExecutionEnvironment
-import org.apache.flink.api.common.functions.GroupReduceFunction
-import org.apache.flink.util.Collector
-import org.apache.flink.examples.java.graph.util.EnumTrianglesData
-import org.apache.flink.api.common.operators.Order
-import scala.collection.mutable.MutableList
-
-import scala.collection.mutable
-
-
-/**
- * Triangle enumeration is a pre-processing step to find closely connected parts in graphs.
- * A triangle consists of three edges that connect three vertices with each other.
- *
- * The basic algorithm works as follows:
- * It groups all edges that share a common vertex and builds triads, i.e., triples of vertices
- * that are connected by two edges. Finally, all triads are filtered for which no third edge exists
- * that closes the triangle.
- *
- * For a group of ''i'' edges that share a common vertex, the number of built triads is quadratic
- * ''(n*(n-1))/2)''. Therefore, an optimization of the algorithm is to group edges on the vertex
- * with the smaller output degree to reduce the number of triads.
- * This implementation extends the basic algorithm by computing output degrees of edge vertices and
- * grouping on edges on the vertex with the smaller degree.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Edges are represented as pairs for vertex IDs which are separated by space
- * characters. Edges are separated by new-line characters.
- * For example `"1 2\n2 12\n1 12\n42 63"` gives four (undirected) edges (1)-(2), (2)-(12),
- * (1)-(12), and (42)-(63) that include a triangle
- *
- * <pre>
- * (1)
- * / \
- * (2)-(12)
- * </pre>
- *
- * Usage:
- * {{{
- * EnumTriangleOpt <edge path> <result path>
- * }}}
- *
- * If no parameters are provided, the program is run with default data from
- * [[org.apache.flink.examples.java.graph.util.EnumTrianglesData]].
- *
- * This example shows how to use:
- *
- * - Custom Java objects which extend Tuple
- * - Group Sorting
- *
- */
-object EnumTrianglesOpt {
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- // set up execution environment
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // read input data
- val edges = getEdgeDataSet(env)
-
- val edgesWithDegrees = edges
- // duplicate and switch edges
- .flatMap(e => Seq(e, Edge(e.v2, e.v1)))
- // add degree of first vertex
- .groupBy("v1").sortGroup("v2", Order.ASCENDING).reduceGroup(new DegreeCounter())
- // join degrees of vertices
- .groupBy("v1", "v2").reduce {
- (e1, e2) =>
- if (e1.d2 == 0) {
- new EdgeWithDegrees(e1.v1, e1.d1, e1.v2, e2.d2)
- } else {
- new EdgeWithDegrees(e1.v1, e2.d1, e1.v2, e1.d2)
- }
- }.withForwardedFields("v1;v2")
-
- // project edges by degrees, vertex with smaller degree comes first
- val edgesByDegree = edgesWithDegrees
- .map(e => if (e.d1 <= e.d2) Edge(e.v1, e.v2) else Edge(e.v2, e.v1))
- // project edges by Id, vertex with smaller Id comes first
- val edgesById = edgesByDegree
- .map(e => if (e.v1 < e.v2) e else Edge(e.v2, e.v1))
-
- val triangles = edgesByDegree
- // build triads
- .groupBy("v1").sortGroup("v2", Order.ASCENDING).reduceGroup(new TriadBuilder())
- // filter triads
- .join(edgesById).where("v2", "v3").equalTo("v1", "v2") { (t, _) => t}
- .withForwardedFieldsFirst("*")
-
- // emit result
- if (fileOutput) {
- triangles.writeAsCsv(outputPath, "\n", ",")
- // execute program
- env.execute("TriangleEnumeration Example")
- } else {
- triangles.print()
- }
-
- }
-
- // *************************************************************************
- // USER DATA TYPES
- // *************************************************************************
-
- case class Edge(v1: Int, v2: Int) extends Serializable
-
- case class Triad(v1: Int, v2: Int, v3: Int) extends Serializable
-
- case class EdgeWithDegrees(v1: Int, d1: Int, v2: Int, d2: Int) extends Serializable
-
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Counts the number of edges that share a common vertex.
- * Emits one edge for each input edge with a degree annotation for the shared vertex.
- * For each emitted edge, the first vertex is the vertex with the smaller id.
- */
- class DegreeCounter extends GroupReduceFunction[Edge, EdgeWithDegrees] {
-
- val vertices = mutable.MutableList[Integer]()
- var groupVertex = 0
-
- override def reduce(edges: java.lang.Iterable[Edge], out: Collector[EdgeWithDegrees]) = {
-
- // empty vertex list
- vertices.clear()
-
- // collect all vertices
- for (e <- edges.asScala) {
- groupVertex = e.v1
- if (!vertices.contains(e.v2) && e.v1 != e.v2) {
- vertices += e.v2
- }
- }
-
- // count vertices to obtain degree of groupVertex
- val degree = vertices.length
-
- // create and emit edges with degrees
- for (v <- vertices) {
- if (v < groupVertex) {
- out.collect(new EdgeWithDegrees(v, 0, groupVertex, degree))
- } else {
- out.collect(new EdgeWithDegrees(groupVertex, degree, v, 0))
- }
- }
- }
- }
-
- /**
- * Builds triads (triples of vertices) from pairs of edges that share a vertex.
- * The first vertex of a triad is the shared vertex, the second and third vertex are ordered by
- * vertexId.
- * Assumes that input edges share the first vertex and are in ascending order of the second
- * vertex.
- */
- @ForwardedFields(Array("v1"))
- class TriadBuilder extends GroupReduceFunction[Edge, Triad] {
-
- val vertices = mutable.MutableList[Integer]()
-
- override def reduce(edges: java.lang.Iterable[Edge], out: Collector[Triad]) = {
-
- // clear vertex list
- vertices.clear()
-
- // build and emit triads
- for (e <- edges.asScala) {
- // combine vertex with all previously read vertices
- for (v <- vertices) {
- out.collect(Triad(e.v1, v, e.v2))
- }
- vertices += e.v2
- }
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- fileOutput = true
- if (args.length == 2) {
- edgePath = args(0)
- outputPath = args(1)
-
- true
- } else {
- System.err.println("Usage: EnumTriangleOpt <edge path> <result path>")
-
- false
- }
- } else {
- System.out.println("Executing Enum Triangles Optimized example with built-in default data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" Usage: EnumTriangleBasic <edge path> <result path>")
-
- true
- }
- }
-
- private def getEdgeDataSet(env: ExecutionEnvironment): DataSet[Edge] = {
- if (fileOutput) {
- env.readCsvFile[Edge](
- edgePath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- } else {
- val edges = EnumTrianglesData.EDGES.map {
- case Array(v1, v2) => new Edge(v1.asInstanceOf[Int], v2.asInstanceOf[Int])}
- env.fromCollection(edges)
- }
- }
-
-
- private var fileOutput: Boolean = false
- private var edgePath: String = null
- private var outputPath: String = null
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/PageRankBasic.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/PageRankBasic.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/PageRankBasic.scala
deleted file mode 100644
index e1d4af6..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/PageRankBasic.scala
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import java.lang.Iterable
-
-import org.apache.flink.api.common.functions.GroupReduceFunction
-import org.apache.flink.api.scala._
-import org.apache.flink.examples.java.graph.util.PageRankData
-import org.apache.flink.api.java.aggregation.Aggregations.SUM
-
-import org.apache.flink.util.Collector
-
-import scala.collection.JavaConverters._
-
-/**
- * A basic implementation of the Page Rank algorithm using a bulk iteration.
- *
- * This implementation requires a set of pages and a set of directed links as input and works as
- * follows.
- *
- * In each iteration, the rank of every page is evenly distributed to all pages it points to. Each
- * page collects the partial ranks of all pages that point to it, sums them up, and applies a
- * dampening factor to the sum. The result is the new rank of the page. A new iteration is started
- * with the new ranks of all pages. This implementation terminates after a fixed number of
- * iterations. This is the Wikipedia entry for the
- * [[http://en.wikipedia.org/wiki/Page_rank Page Rank algorithm]]
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Pages represented as an (long) ID separated by new-line characters.
- * For example `"1\n2\n12\n42\n63"` gives five pages with IDs 1, 2, 12, 42, and 63.
- * - Links are represented as pairs of page IDs which are separated by space characters. Links
- * are separated by new-line characters.
- * For example `"1 2\n2 12\n1 12\n42 63"` gives four (directed) links (1)->(2), (2)->(12),
- * (1)->(12), and (42)->(63). For this simple implementation it is required that each page has
- * at least one incoming and one outgoing link (a page can point to itself).
- *
- * Usage:
- * {{{
- * PageRankBasic <pages path> <links path> <output path> <num pages> <num iterations>
- * }}}
- *
- * If no parameters are provided, the program is run with default data from
- * [[org.apache.flink.examples.java.graph.util.PageRankData]] and 10 iterations.
- *
- * This example shows how to use:
- *
- * - Bulk Iterations
- * - Default Join
- * - Configure user-defined functions using constructor parameters.
- *
- */
-object PageRankBasic {
-
- private final val DAMPENING_FACTOR: Double = 0.85
- private final val EPSILON: Double = 0.0001
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- // set up execution environment
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // read input data
- val pages = getPagesDataSet(env)
- val links = getLinksDataSet(env)
-
- // assign initial ranks to pages
- val pagesWithRanks = pages.map(p => Page(p, 1.0 / numPages)).withForwardedFields("*->pageId")
-
- // build adjacency list from link input
- val adjacencyLists = links
- .groupBy("sourceId").reduceGroup( new GroupReduceFunction[Link, AdjacencyList] {
- override def reduce(values: Iterable[Link], out: Collector[AdjacencyList]): Unit = {
- var outputId = -1L
- val outputList = values.asScala map { t => outputId = t.sourceId; t.targetId }
- out.collect(new AdjacencyList(outputId, outputList.toArray))
- }
- })
-
- // start iteration
- val finalRanks = pagesWithRanks.iterateWithTermination(maxIterations) {
- currentRanks =>
- val newRanks = currentRanks
- // distribute ranks to target pages
- .join(adjacencyLists).where("pageId").equalTo("sourceId") {
- (page, adjacent, out: Collector[Page]) =>
- val targets = adjacent.targetIds
- val len = targets.length
- adjacent.targetIds foreach { t => out.collect(Page(t, page.rank /len )) }
- }
- // collect ranks and sum them up
- .groupBy("pageId").aggregate(SUM, "rank")
- // apply dampening factor
- .map { p =>
- Page(p.pageId, (p.rank * DAMPENING_FACTOR) + ((1 - DAMPENING_FACTOR) / numPages))
- }.withForwardedFields("pageId")
-
- // terminate if no rank update was significant
- val termination = currentRanks.join(newRanks).where("pageId").equalTo("pageId") {
- (current, next, out: Collector[Int]) =>
- // check for significant update
- if (math.abs(current.rank - next.rank) > EPSILON) out.collect(1)
- }
- (newRanks, termination)
- }
-
- val result = finalRanks
-
- // emit result
- if (fileOutput) {
- result.writeAsCsv(outputPath, "\n", " ")
- // execute program
- env.execute("Basic PageRank Example")
- } else {
- result.print()
- }
- }
-
- // *************************************************************************
- // USER TYPES
- // *************************************************************************
-
- case class Link(sourceId: Long, targetId: Long)
-
- case class Page(pageId: Long, rank: Double)
-
- case class AdjacencyList(sourceId: Long, targetIds: Array[Long])
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- fileOutput = true
- if (args.length == 5) {
- pagesInputPath = args(0)
- linksInputPath = args(1)
- outputPath = args(2)
- numPages = args(3).toLong
- maxIterations = args(4).toInt
-
- true
- } else {
- System.err.println("Usage: PageRankBasic <pages path> <links path> <output path> <num " +
- "pages> <num iterations>")
-
- false
- }
- } else {
- System.out.println("Executing PageRank Basic example with default parameters and built-in " +
- "default data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" Usage: PageRankBasic <pages path> <links path> <output path> <num " +
- "pages> <num iterations>")
-
- numPages = PageRankData.getNumberOfPages
-
- true
- }
- }
-
- private def getPagesDataSet(env: ExecutionEnvironment): DataSet[Long] = {
- if (fileOutput) {
- env.readCsvFile[Tuple1[Long]](pagesInputPath, fieldDelimiter = " ", lineDelimiter = "\n")
- .map(x => x._1)
- } else {
- env.generateSequence(1, 15)
- }
- }
-
- private def getLinksDataSet(env: ExecutionEnvironment): DataSet[Link] = {
- if (fileOutput) {
- env.readCsvFile[Link](linksInputPath, fieldDelimiter = " ",
- includedFields = Array(0, 1))
- } else {
- val edges = PageRankData.EDGES.map { case Array(v1, v2) => Link(v1.asInstanceOf[Long],
- v2.asInstanceOf[Long])}
- env.fromCollection(edges)
- }
- }
-
- private var fileOutput: Boolean = false
- private var pagesInputPath: String = null
- private var linksInputPath: String = null
- private var outputPath: String = null
- private var numPages: Long = 0
- private var maxIterations: Int = 10
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/TransitiveClosureNaive.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/TransitiveClosureNaive.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/TransitiveClosureNaive.scala
deleted file mode 100644
index 3de0f2e..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/graph/TransitiveClosureNaive.scala
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.flink.examples.scala.graph
-
-import org.apache.flink.api.scala._
-import org.apache.flink.examples.java.graph.util.ConnectedComponentsData
-import org.apache.flink.util.Collector
-
-object TransitiveClosureNaive {
-
- def main (args: Array[String]): Unit = {
- if (!parseParameters(args)) {
- return
- }
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- val edges = getEdgesDataSet(env)
-
- val paths = edges.iterateWithTermination(maxIterations) { prevPaths: DataSet[(Long, Long)] =>
-
- val nextPaths = prevPaths
- .join(edges)
- .where(1).equalTo(0) {
- (left, right) => (left._1,right._2)
- }.withForwardedFieldsFirst("_1").withForwardedFieldsSecond("_2")
- .union(prevPaths)
- .groupBy(0, 1)
- .reduce((l, r) => l).withForwardedFields("_1; _2")
-
- val terminate = prevPaths
- .coGroup(nextPaths)
- .where(0).equalTo(0) {
- (prev, next, out: Collector[(Long, Long)]) => {
- val prevPaths = prev.toSet
- for (n <- next)
- if (!prevPaths.contains(n)) out.collect(n)
- }
- }.withForwardedFieldsSecond("*")
- (nextPaths, terminate)
- }
-
- if (fileOutput) {
- paths.writeAsCsv(outputPath, "\n", " ")
- env.execute("Scala Transitive Closure Example")
- } else {
- paths.print()
- }
-
-
-
- }
-
-
- private var fileOutput: Boolean = false
- private var edgesPath: String = null
- private var outputPath: String = null
- private var maxIterations: Int = 10
-
- private def parseParameters(programArguments: Array[String]): Boolean = {
- if (programArguments.length > 0) {
- fileOutput = true
- if (programArguments.length == 3) {
- edgesPath = programArguments(0)
- outputPath = programArguments(1)
- maxIterations = Integer.parseInt(programArguments(2))
- }
- else {
- System.err.println("Usage: TransitiveClosure <edges path> <result path> <max number of " +
- "iterations>")
- return false
- }
- }
- else {
- System.out.println("Executing TransitiveClosure example with default parameters and " +
- "built-in default data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" Usage: TransitiveClosure <edges path> <result path> <max number of " +
- "iterations>")
- }
- true
- }
-
- private def getEdgesDataSet(env: ExecutionEnvironment): DataSet[(Long, Long)] = {
- if (fileOutput) {
- env.readCsvFile[(Long, Long)](
- edgesPath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- .map { x => (x._1, x._2)}
- }
- else {
- val edgeData = ConnectedComponentsData.EDGES map {
- case Array(x, y) => (x.asInstanceOf[Long], y.asInstanceOf[Long])
- }
- env.fromCollection(edgeData)
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/misc/PiEstimation.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/misc/PiEstimation.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/misc/PiEstimation.scala
deleted file mode 100644
index 3453ee8..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/misc/PiEstimation.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.examples.scala.misc
-
-import org.apache.flink.api.scala._
-
-object PiEstimation {
-
- def main(args: Array[String]) {
-
- val numSamples: Long = if (args.length > 0) args(0).toLong else 1000000
-
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // count how many of the samples would randomly fall into
- // the upper right quadrant of the unit circle
- val count =
- env.generateSequence(1, numSamples)
- .map { sample =>
- val x = Math.random()
- val y = Math.random()
- if (x * x + y * y < 1) 1L else 0L
- }
- .reduce(_ + _)
-
- // ratio of samples in upper right quadrant vs total samples gives surface of upper
- // right quadrant, times 4 gives surface of whole unit circle, i.e. PI
- val pi = count
- .map ( _ * 4.0 / numSamples)
-
- println("We estimate Pi to be:")
-
- pi.print()
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/d0e1d635/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala
----------------------------------------------------------------------
diff --git a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala b/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala
deleted file mode 100644
index 2a7b786..0000000
--- a/flink-examples/flink-scala-examples/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala
+++ /dev/null
@@ -1,195 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.examples.scala.ml
-
-import org.apache.flink.api.common.functions._
-import org.apache.flink.api.scala._
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.examples.java.ml.util.LinearRegressionData
-
-import scala.collection.JavaConverters._
-
-/**
- * This example implements a basic Linear Regression to solve the y = theta0 + theta1*x problem
- * using batch gradient descent algorithm.
- *
- * Linear Regression with BGD(batch gradient descent) algorithm is an iterative algorithm and
- * works as follows:
- *
- * Giving a data set and target set, the BGD try to find out the best parameters for the data set
- * to fit the target set.
- * In each iteration, the algorithm computes the gradient of the cost function and use it to
- * update all the parameters.
- * The algorithm terminates after a fixed number of iterations (as in this implementation).
- * With enough iteration, the algorithm can minimize the cost function and find the best parameters
- * This is the Wikipedia entry for the
- * [[http://en.wikipedia.org/wiki/Linear_regression Linear regression]] and
- * [[http://en.wikipedia.org/wiki/Gradient_descent Gradient descent algorithm]].
- *
- * This implementation works on one-dimensional data and finds the best two-dimensional theta to
- * fit the target.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Data points are represented as two double values separated by a blank character. The first
- * one represent the X(the training data) and the second represent the Y(target). Data points are
- * separated by newline characters.
- * For example `"-0.02 -0.04\n5.3 10.6\n"`gives two data points
- * (x=-0.02, y=-0.04) and (x=5.3, y=10.6).
- *
- * This example shows how to use:
- *
- * - Bulk iterations
- * - Broadcast variables in bulk iterations
- */
-object LinearRegression {
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- val env = ExecutionEnvironment.getExecutionEnvironment
- val data = getDataSet(env)
- val parameters = getParamsDataSet(env)
-
- val result = parameters.iterate(numIterations) { currentParameters =>
- val newParameters = data
- .map(new SubUpdate).withBroadcastSet(currentParameters, "parameters")
- .reduce { (p1, p2) =>
- val result = p1._1 + p2._1
- (result, p1._2 + p2._2)
- }
- .map { x => x._1.div(x._2) }
- newParameters
- }
-
- if (fileOutput) {
- result.writeAsText(outputPath)
- env.execute("Scala Linear Regression example")
- }
- else {
- result.print()
- }
- }
-
- /**
- * A simple data sample, x means the input, and y means the target.
- */
- case class Data(var x: Double, var y: Double)
-
- /**
- * A set of parameters -- theta0, theta1.
- */
- case class Params(theta0: Double, theta1: Double) {
- def div(a: Int): Params = {
- Params(theta0 / a, theta1 / a)
- }
-
- def + (other: Params) = {
- Params(theta0 + other.theta0, theta1 + other.theta1)
- }
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Compute a single BGD type update for every parameters.
- */
- class SubUpdate extends RichMapFunction[Data, (Params, Int)] {
-
- private var parameter: Params = null
-
- /** Reads the parameters from a broadcast variable into a collection. */
- override def open(parameters: Configuration) {
- val parameters = getRuntimeContext.getBroadcastVariable[Params]("parameters").asScala
- parameter = parameters.head
- }
-
- def map(in: Data): (Params, Int) = {
- val theta0 =
- parameter.theta0 - 0.01 * ((parameter.theta0 + (parameter.theta1 * in.x)) - in.y)
- val theta1 =
- parameter.theta1 - 0.01 * (((parameter.theta0 + (parameter.theta1 * in.x)) - in.y) * in.x)
- (Params(theta0, theta1), 1)
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
- private var fileOutput: Boolean = false
- private var dataPath: String = null
- private var outputPath: String = null
- private var numIterations: Int = 10
-
- private def parseParameters(programArguments: Array[String]): Boolean = {
- if (programArguments.length > 0) {
- fileOutput = true
- if (programArguments.length == 3) {
- dataPath = programArguments(0)
- outputPath = programArguments(1)
- numIterations = programArguments(2).toInt
-
- true
- }
- else {
- System.err.println("Usage: LinearRegression <data path> <result path> <num iterations>")
-
- false
- }
- }
- else {
- System.out.println("Executing Linear Regression example with default parameters and " +
- "built-in default data.")
- System.out.println(" Provide parameters to read input data from files.")
- System.out.println(" See the documentation for the correct format of input files.")
- System.out.println(" We provide a data generator to create synthetic input files for this " +
- "program.")
- System.out.println(" Usage: LinearRegression <data path> <result path> <num iterations>")
-
- true
- }
- }
-
- private def getDataSet(env: ExecutionEnvironment): DataSet[Data] = {
- if (fileOutput) {
- env.readCsvFile[(Double, Double)](
- dataPath,
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- .map { t => new Data(t._1, t._2) }
- }
- else {
- val data = LinearRegressionData.DATA map {
- case Array(x, y) => Data(x.asInstanceOf[Double], y.asInstanceOf[Double])
- }
- env.fromCollection(data)
- }
- }
-
- private def getParamsDataSet(env: ExecutionEnvironment): DataSet[Params] = {
- val params = LinearRegressionData.PARAMS map {
- case Array(x, y) => Params(x.asInstanceOf[Double], y.asInstanceOf[Double])
- }
- env.fromCollection(params)
- }
-}