You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/07/08 11:05:42 UTC
[1/3] flink git commit: [FLINK-2293] [runtime] Fix estimation for the
number of hash buckets on recursive builds
Repository: flink
Updated Branches:
refs/heads/master 0d2c49005 -> 9ea4be895
[FLINK-2293] [runtime] Fix estimation for the number of hash buckets on recursive builds
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/627f3cbc
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/627f3cbc
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/627f3cbc
Branch: refs/heads/master
Commit: 627f3cbcfdca8368eea6aa825cd9a45a9a0a841f
Parents: 0d2c490
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jul 7 17:01:44 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 7 17:01:44 2015 +0200
----------------------------------------------------------------------
.../operators/hash/MutableHashTable.java | 23 ++++++++++----------
1 file changed, 11 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/627f3cbc/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
index 21d67a8..9416796 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/hash/MutableHashTable.java
@@ -678,9 +678,7 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
* @param input
* @throws IOException
*/
- protected void buildInitialTable(final MutableObjectIterator<BT> input)
- throws IOException
- {
+ protected void buildInitialTable(final MutableObjectIterator<BT> input) throws IOException {
// create the partitions
final int partitionFanOut = getPartitioningFanOutNoEstimates(this.availableMemory.size());
if (partitionFanOut > MAX_NUM_PARTITIONS) {
@@ -788,8 +786,8 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
final int avgRecordLenPartition = (int) (((long) p.getBuildSideBlockCount()) *
this.segmentSize / p.getBuildSideRecordCount());
- final int bucketCount = (int) (((long) totalBuffersAvailable) * RECORD_TABLE_BYTES /
- (avgRecordLenPartition + RECORD_OVERHEAD_BYTES));
+ final int bucketCount = getInitialTableSize(totalBuffersAvailable, this.segmentSize,
+ getPartitioningFanOutNoEstimates(totalBuffersAvailable), avgRecordLenPartition);
// compute in how many splits, we'd need to partition the result
final int splits = (int) (totalBuffersNeeded / totalBuffersAvailable) + 1;
@@ -1201,7 +1199,7 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
* @param numBuffers The number of available buffers.
* @return The number
*/
- public static final int getNumWriteBehindBuffers(int numBuffers) {
+ public static int getNumWriteBehindBuffers(int numBuffers) {
int numIOBufs = (int) (Math.log(numBuffers) / Math.log(4) - 1.5);
return numIOBufs > 6 ? 6 : numIOBufs;
}
@@ -1216,11 +1214,12 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
* @param numBuffers The number of buffers available.
* @return The number of partitions to use.
*/
- public static final int getPartitioningFanOutNoEstimates(int numBuffers) {
+ public static int getPartitioningFanOutNoEstimates(int numBuffers) {
return Math.max(10, Math.min(numBuffers / 10, MAX_NUM_PARTITIONS));
}
- public static final int getInitialTableSize(int numBuffers, int bufferSize, int numPartitions, int recordLenBytes) {
+ public static int getInitialTableSize(int numBuffers, int bufferSize, int numPartitions, int recordLenBytes) {
+
// ----------------------------------------------------------------------------------------
// the following observations hold:
// 1) If the records are assumed to be very large, then many buffers need to go to the partitions
@@ -1249,11 +1248,11 @@ public class MutableHashTable<BT, PT> implements MemorySegmentSource {
/**
* Assigns a partition to a bucket.
*
- * @param bucket
- * @param numPartitions
- * @return The hash code for the integer.
+ * @param bucket The bucket to get the partition for.
+ * @param numPartitions The number of partitions.
+ * @return The partition for the bucket.
*/
- public static final byte assignPartition(int bucket, byte numPartitions) {
+ public static byte assignPartition(int bucket, byte numPartitions) {
return (byte) (bucket % numPartitions);
}
[3/3] flink git commit: [FLINK-2000] [table api] Add sql style
aggregation support
Posted by se...@apache.org.
[FLINK-2000] [table api] Add sql style aggregation support
This commit exists only to trigger closing the pull request.
This closes #782
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9ea4be89
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9ea4be89
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9ea4be89
Branch: refs/heads/master
Commit: 9ea4be8957be0fdcb388315425d4d72775d71951
Parents: 474f9c6
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Jul 7 17:05:22 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 7 17:05:22 2015 +0200
----------------------------------------------------------------------
----------------------------------------------------------------------
[2/3] flink git commit: [FLINK-2242] [api] Deprecate RemoteCollector
interface and implementation
Posted by se...@apache.org.
[FLINK-2242] [api] Deprecate RemoteCollector interface and implementation
This closes #852
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/474f9c65
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/474f9c65
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/474f9c65
Branch: refs/heads/master
Commit: 474f9c65cd4cd11423aac9261ed904378c31f12b
Parents: 627f3cb
Author: Maximilian Michels <mx...@apache.org>
Authored: Fri Jun 19 10:39:27 2015 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Jul 7 17:03:05 2015 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/api/java/io/RemoteCollector.java | 6 ++++++
.../java/org/apache/flink/api/java/io/RemoteCollectorImpl.java | 4 ++++
.../apache/flink/api/java/io/RemoteCollectorOutputFormat.java | 5 +++++
3 files changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/474f9c65/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollector.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollector.java b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollector.java
index 2b3e575..bcfc332 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollector.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollector.java
@@ -18,6 +18,8 @@
package org.apache.flink.api.java.io;
+import org.apache.flink.api.java.DataSet;
+
import java.rmi.Remote;
import java.rmi.RemoteException;
@@ -27,7 +29,11 @@ import java.rmi.RemoteException;
*
* @param <T>
* The type of the records the collector will receive
+ *
+ * @deprecated Results are retrieved through {@link org.apache.flink.api.common.accumulators.Accumulator}
+ * and the {@link DataSet#collect()} method respectively.
*/
+@Deprecated
public interface RemoteCollector<T> extends Remote {
public void collect(T element) throws RemoteException;
http://git-wip-us.apache.org/repos/asf/flink/blob/474f9c65/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorImpl.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorImpl.java b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorImpl.java
index f30905b..2d080ab 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorImpl.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorImpl.java
@@ -44,8 +44,12 @@ import org.apache.flink.api.java.operators.DataSink;
/**
* This class provides a counterpart implementation for the
* {@link RemoteCollectorOutputFormat}.
+ *
+ * @deprecated Results are retrieved through {@link org.apache.flink.api.common.accumulators.Accumulator}
+ * and the {@link DataSet#collect()} method respectively.
*/
+@Deprecated
public class RemoteCollectorImpl<T> extends UnicastRemoteObject implements
RemoteCollector<T> {
http://git-wip-us.apache.org/repos/asf/flink/blob/474f9c65/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorOutputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorOutputFormat.java
index 06def88..3fe5cef 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorOutputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/io/RemoteCollectorOutputFormat.java
@@ -26,6 +26,7 @@ import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.DataSet;
import org.apache.flink.configuration.Configuration;
/**
@@ -35,7 +36,11 @@ import org.apache.flink.configuration.Configuration;
* into an instance of {@link RemoteCollectorOutputFormat}. Further in the
* client's VM parameters -Djava.rmi.server.hostname should be set to the own IP
* address.
+ *
+ * @deprecated Results are retrieved through {@link org.apache.flink.api.common.accumulators.Accumulator}
+ * and the {@link DataSet#collect()} method respectively.
*/
+@Deprecated
public class RemoteCollectorOutputFormat<T> implements OutputFormat<T> {
private static final long serialVersionUID = 1922744224032398102L;