You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2017/12/04 07:52:39 UTC
[1/9] ignite git commit: IGNITE-6997 .NET: Thin client: Move
ClientStatus to public API
Repository: ignite
Updated Branches:
refs/heads/ignite-zk 73cea1642 -> 74330bc14
IGNITE-6997 .NET: Thin client: Move ClientStatus to public API
This closes #3120
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3979e6a9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3979e6a9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3979e6a9
Branch: refs/heads/ignite-zk
Commit: 3979e6a9b561f3ada2ca0e108c76885725ebacb6
Parents: 23b106d
Author: Pavel Tupitsyn <pt...@apache.org>
Authored: Fri Dec 1 14:24:49 2017 +0300
Committer: Pavel Tupitsyn <pt...@apache.org>
Committed: Fri Dec 1 14:24:49 2017 +0300
----------------------------------------------------------------------
.../Client/Cache/CacheTest.cs | 4 +-
.../Client/Cache/CreateCacheTest.cs | 4 +-
.../Client/Cache/ScanQueryTest.cs | 4 +-
.../Client/ClientConnectionTest.cs | 2 +-
.../Client/RawSocketTest.cs | 3 +-
.../Apache.Ignite.Core.csproj | 2 +-
.../Client/ClientStatusCode.cs | 57 ++++++++++++++++++++
.../Client/IgniteClientException.cs | 24 ++++-----
.../Impl/Client/Cache/CacheClient.cs | 8 +--
.../Impl/Client/ClientSocket.cs | 8 +--
.../Impl/Client/ClientStatus.cs | 32 -----------
11 files changed, 85 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheTest.cs
index 0d82479..f54efae 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CacheTest.cs
@@ -840,9 +840,7 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
var ex = Assert.Throws<IgniteClientException>(() => cache.Put(1, 1));
Assert.AreEqual("Cache doesn't exist: foobar", ex.Message);
-#if !NETCOREAPP2_0
- Assert.AreEqual((int) Impl.Client.ClientStatus.CacheDoesNotExist, ex.ErrorCode);
-#endif
+ Assert.AreEqual(ClientStatusCode.CacheDoesNotExist, ex.StatusCode);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CreateCacheTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CreateCacheTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CreateCacheTest.cs
index 5e2ccf4..4489462 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CreateCacheTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/CreateCacheTest.cs
@@ -87,7 +87,7 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
var ex = Assert.Throws<IgniteClientException>(() => Client.CreateCache<int, int>(cache.Name));
Assert.AreEqual(
"Failed to start cache (a cache with the same name is already started): foobar", ex.Message);
- Assert.AreEqual((int) ClientStatus.CacheExists, ex.ErrorCode);
+ Assert.AreEqual(ClientStatusCode.CacheExists, ex.StatusCode);
// Template: custom configuration.
cache = Client.CreateCache<int, int>(TemplateCacheName.Replace("*", "1"));
@@ -140,7 +140,7 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
var ex = Assert.Throws<IgniteClientException>(() => Client.CreateCache<int, int>(cfg));
Assert.AreEqual(
"Failed to start cache (a cache with the same name is already started): a", ex.Message);
- Assert.AreEqual((int) ClientStatus.CacheExists, ex.ErrorCode);
+ Assert.AreEqual(ClientStatusCode.CacheExists, ex.StatusCode);
// Custom config.
cfg = GetFullCacheConfiguration("b");
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ScanQueryTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ScanQueryTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ScanQueryTest.cs
index 354b869..71c8f0f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ScanQueryTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/Cache/ScanQueryTest.cs
@@ -203,9 +203,7 @@ namespace Apache.Ignite.Core.Tests.Client.Cache
// MaxCursors = 3
var ex = Assert.Throws<IgniteClientException>(() => clientCache.Query(qry));
Assert.AreEqual("Too many open cursors", ex.Message.Substring(0, 21));
-#if !NETCOREAPP2_0
- Assert.AreEqual((int) Impl.Client.ClientStatus.TooManyCursors, ex.ErrorCode);
-#endif
+ Assert.AreEqual(ClientStatusCode.TooManyCursors, ex.StatusCode);
var count = 0;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
index 4b9af70..6e368a3 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/ClientConnectionTest.cs
@@ -124,7 +124,7 @@ namespace Apache.Ignite.Core.Tests.Client
new Impl.Client.ClientSocket(GetClientConfiguration(),
new Impl.Client.ClientProtocolVersion(-1, -1, -1)));
- Assert.AreEqual((int) Impl.Client.ClientStatus.Fail, ex.ErrorCode);
+ Assert.AreEqual(ClientStatusCode.Fail, ex.StatusCode);
Assert.AreEqual("Client handhsake failed: 'Unsupported version.'. " +
"Client version: -1.-1.-1. Server version: 1.0.0", ex.Message);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/RawSocketTest.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/RawSocketTest.cs b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/RawSocketTest.cs
index 9aab341..b088bb6 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/RawSocketTest.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core.Tests/Client/RawSocketTest.cs
@@ -20,6 +20,7 @@ namespace Apache.Ignite.Core.Tests.Client
using System;
using System.Net;
using System.Net.Sockets;
+ using Apache.Ignite.Core.Client;
using Apache.Ignite.Core.Configuration;
using Apache.Ignite.Core.Impl;
using Apache.Ignite.Core.Impl.Binary;
@@ -106,7 +107,7 @@ namespace Apache.Ignite.Core.Tests.Client
Assert.AreEqual(11, requestId);
var status = reader.ReadInt();
- Assert.AreEqual((int) ClientStatus.InvalidOpCode, status);
+ Assert.AreEqual((int) ClientStatusCode.InvalidOpCode, status);
var err = reader.ReadObject<string>();
Assert.AreEqual("Invalid request op code: -1", err);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
index 5adb501..2852853 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Apache.Ignite.Core.csproj
@@ -129,7 +129,7 @@
<Compile Include="Impl\Binary\IBinaryProcessor.cs" />
<Compile Include="Impl\Client\Cache\Query\ClientQueryCursorBase.cs" />
<Compile Include="Impl\Client\Cache\Query\StatementType.cs" />
- <Compile Include="Impl\Client\ClientStatus.cs" />
+ <Compile Include="Client\ClientStatusCode.cs" />
<Compile Include="Events\LocalEventListener.cs" />
<Compile Include="Impl\DataStorageMetrics.cs" />
<Compile Include="Impl\IIgniteInternal.cs" />
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
new file mode 100644
index 0000000..3f5ee8e
--- /dev/null
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/ClientStatusCode.cs
@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+namespace Apache.Ignite.Core.Client
+{
+ using Apache.Ignite.Core.Configuration;
+
+ /// <summary>
+ /// Client status codes (see <see cref="IgniteClientException.StatusCode"/>).
+ /// </summary>
+ public enum ClientStatusCode
+ {
+ /// <summary>
+ /// Operation succeeded.
+ /// </summary>
+ Success = 0,
+
+ /// <summary>
+ /// Operation failed (general-purpose code).
+ /// </summary>
+ Fail = 1,
+
+ /// <summary>
+ /// Invalid request operation code.
+ /// </summary>
+ InvalidOpCode = 2,
+
+ /// <summary>
+ /// Specified cache does not exist.
+ /// </summary>
+ CacheDoesNotExist = 1000,
+
+ /// <summary>
+ /// Cache already exists.
+ /// </summary>
+ CacheExists = 1001,
+
+ /// <summary>
+ /// The too many cursors (see <see cref="ClientConnectorConfiguration.MaxOpenCursorsPerConnection"/>).
+ /// </summary>
+ TooManyCursors = 1010
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Client/IgniteClientException.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Client/IgniteClientException.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Client/IgniteClientException.cs
index 2df3d1b..3165d45 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Client/IgniteClientException.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Client/IgniteClientException.cs
@@ -20,7 +20,6 @@ namespace Apache.Ignite.Core.Client
using System;
using System.Runtime.Serialization;
using Apache.Ignite.Core.Common;
- using Apache.Ignite.Core.Impl.Client;
/// <summary>
/// Ignite thin client exception.
@@ -29,17 +28,17 @@ namespace Apache.Ignite.Core.Client
public class IgniteClientException : IgniteException
{
/** Error code field. */
- private const string ErrorCodeField = "ErrorCode";
+ private const string ErrorCodeField = "StatusCode";
/** Error code. */
- private readonly int _errorCode = (int) ClientStatus.Fail;
+ private readonly ClientStatusCode _statusCode = ClientStatusCode.Fail;
/// <summary>
- /// Gets the error code.
+ /// Gets the status code code.
/// </summary>
- public int ErrorCode
+ public ClientStatusCode StatusCode
{
- get { return _errorCode; }
+ get { return _statusCode; }
}
/// <summary>
@@ -74,10 +73,11 @@ namespace Apache.Ignite.Core.Client
/// </summary>
/// <param name="message">The message.</param>
/// <param name="cause">The cause.</param>
- /// <param name="errorCode">The error code.</param>
- public IgniteClientException(string message, Exception cause, int errorCode) : base(message, cause)
+ /// <param name="statusCode">The error code.</param>
+ public IgniteClientException(string message, Exception cause, ClientStatusCode statusCode)
+ : base(message, cause)
{
- _errorCode = errorCode;
+ _statusCode = statusCode;
}
/// <summary>
@@ -87,7 +87,7 @@ namespace Apache.Ignite.Core.Client
/// <param name="ctx">Streaming context.</param>
protected IgniteClientException(SerializationInfo info, StreamingContext ctx) : base(info, ctx)
{
- _errorCode = info.GetInt32(ErrorCodeField);
+ _statusCode = (ClientStatusCode) info.GetInt32(ErrorCodeField);
}
/// <summary>
@@ -102,7 +102,7 @@ namespace Apache.Ignite.Core.Client
{
base.GetObjectData(info, context);
- info.AddValue(ErrorCodeField, _errorCode);
+ info.AddValue(ErrorCodeField, (int) _statusCode);
}
/// <summary>
@@ -110,7 +110,7 @@ namespace Apache.Ignite.Core.Client
/// </summary>
public override string ToString()
{
- return string.Format("{0} [ErrorCode={1}]", base.ToString(), ErrorCode);
+ return string.Format("{0} [StatusCode={1}]", base.ToString(), StatusCode);
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
index 0ed2c6d..45c0b0f 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/Cache/CacheClient.cs
@@ -522,15 +522,15 @@ namespace Apache.Ignite.Core.Impl.Client.Cache
/// <summary>
/// Handles the error.
/// </summary>
- private T HandleError<T>(ClientStatus status, string msg)
+ private T HandleError<T>(ClientStatusCode status, string msg)
{
switch (status)
{
- case ClientStatus.CacheDoesNotExist:
- throw new IgniteClientException("Cache doesn't exist: " + Name, null, (int) status);
+ case ClientStatusCode.CacheDoesNotExist:
+ throw new IgniteClientException("Cache doesn't exist: " + Name, null, status);
default:
- throw new IgniteClientException(msg, null, (int) status);
+ throw new IgniteClientException(msg, null, status);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
index 078927b..b8218c1 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientSocket.cs
@@ -68,7 +68,7 @@ namespace Apache.Ignite.Core.Impl.Client
/// Performs a send-receive operation.
/// </summary>
public T DoOutInOp<T>(ClientOp opId, Action<IBinaryStream> writeAction,
- Func<IBinaryStream, T> readFunc, Func<ClientStatus, string, T> errorFunc = null)
+ Func<IBinaryStream, T> readFunc, Func<ClientStatusCode, string, T> errorFunc = null)
{
var requestId = Interlocked.Increment(ref _requestId);
@@ -88,9 +88,9 @@ namespace Apache.Ignite.Core.Impl.Client
var resRequestId = stream.ReadLong();
Debug.Assert(requestId == resRequestId);
- var statusCode = (ClientStatus) stream.ReadInt();
+ var statusCode = (ClientStatusCode) stream.ReadInt();
- if (statusCode == ClientStatus.Success)
+ if (statusCode == ClientStatusCode.Success)
{
return readFunc != null ? readFunc(stream) : default(T);
}
@@ -102,7 +102,7 @@ namespace Apache.Ignite.Core.Impl.Client
return errorFunc(statusCode, msg);
}
- throw new IgniteClientException(msg, null, (int) statusCode);
+ throw new IgniteClientException(msg, null, statusCode);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3979e6a9/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientStatus.cs
----------------------------------------------------------------------
diff --git a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientStatus.cs b/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientStatus.cs
deleted file mode 100644
index 294a202..0000000
--- a/modules/platforms/dotnet/Apache.Ignite.Core/Impl/Client/ClientStatus.cs
+++ /dev/null
@@ -1,32 +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.
- */
-
-namespace Apache.Ignite.Core.Impl.Client
-{
- /// <summary>
- /// Client status codes.
- /// </summary>
- internal enum ClientStatus
- {
- Success = 0,
- Fail = 1,
- InvalidOpCode = 2,
- CacheDoesNotExist = 1000,
- CacheExists = 1001,
- TooManyCursors = 1010
- }
-}
[7/9] ignite git commit: IGNITE-7091 Fixing assertion raising because
of serializer switch - Fixes #3128.
Posted by sb...@apache.org.
IGNITE-7091 Fixing assertion raising because of serializer switch - Fixes #3128.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/6101fde4
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/6101fde4
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/6101fde4
Branch: refs/heads/ignite-zk
Commit: 6101fde4dc73431bf89cd01dfd828a10adad13ef
Parents: b13f088
Author: dpavlov <dp...@gridgain.com>
Authored: Mon Dec 4 09:31:56 2017 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Dec 4 09:31:56 2017 +0300
----------------------------------------------------------------------
.../DummyPersistenceCompatibilityTest.java | 2 +-
.../wal/FileWriteAheadLogManager.java | 27 ++++++++++++++------
2 files changed, 20 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/6101fde4/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/DummyPersistenceCompatibilityTest.java
----------------------------------------------------------------------
diff --git a/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/DummyPersistenceCompatibilityTest.java b/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/DummyPersistenceCompatibilityTest.java
index 466b858..64a1cbf 100644
--- a/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/DummyPersistenceCompatibilityTest.java
+++ b/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/DummyPersistenceCompatibilityTest.java
@@ -44,7 +44,7 @@ import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
*/
public class DummyPersistenceCompatibilityTest extends IgnitePersistenceCompatibilityAbstractTest {
/** */
- private static final String TEST_CACHE_NAME = DummyPersistenceCompatibilityTest.class.getSimpleName();
+ protected static final String TEST_CACHE_NAME = DummyPersistenceCompatibilityTest.class.getSimpleName();
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
http://git-wip-us.apache.org/repos/asf/ignite/blob/6101fde4/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
index 948a8ae..5cb6b2f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/FileWriteAheadLogManager.java
@@ -1381,13 +1381,19 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/**
* @param absIdx Segment absolute index.
- * @return {@code True} if can read, {@code false} if work segment
+ * @return <ul><li>{@code True} if can read, no lock is held, </li><li>{@code false} if work segment, need
+ * release segment later, use {@link #releaseWorkSegment} for unlock</li> </ul>
*/
@SuppressWarnings("NonPrivateFieldAccessedInSynchronizedContext")
private boolean checkCanReadArchiveOrReserveWorkSegment(long absIdx) {
synchronized (this) {
- if (lastAbsArchivedIdx >= absIdx)
+ if (lastAbsArchivedIdx >= absIdx) {
+ if (log.isDebugEnabled())
+ log.debug("Not needed to reserve WAL segment: absIdx=" + absIdx + ";" +
+ " lastAbsArchivedIdx=" + lastAbsArchivedIdx);
+
return true;
+ }
Integer cur = locked.get(absIdx);
@@ -1410,7 +1416,8 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
synchronized (this) {
Integer cur = locked.get(absIdx);
- assert cur != null && cur > 0;
+ assert cur != null && cur > 0 : "WAL Segment with Index " + absIdx + " is not locked;" +
+ " lastAbsArchivedIdx = " + lastAbsArchivedIdx;
if (cur == 1) {
locked.remove(absIdx);
@@ -3070,10 +3077,13 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
nextHandle = null;
}
- if (nextHandle != null)
- nextHandle.workDir = !readArchive;
+ if (nextHandle == null) {
+ if (!readArchive)
+ releaseWorkSegment(curWalSegmIdx);
+ }
else
- releaseWorkSegment(curWalSegmIdx);
+ nextHandle.workDir = !readArchive;
+
curRec = null;
return nextHandle;
@@ -3081,8 +3091,9 @@ public class FileWriteAheadLogManager extends GridCacheSharedManagerAdapter impl
/**
* @param absIdx Absolute index to check.
- * @return {@code True} if we can safely read the archive, {@code false} if the segment has not been archived
- * yet. In this case the corresponding work segment is reserved (will not be deleted until release).
+ * @return <ul><li> {@code True} if we can safely read the archive, </li> <li>{@code false} if the segment has
+ * not been archived yet. In this case the corresponding work segment is reserved (will not be deleted until
+ * release). Use {@link #releaseWorkSegment} for unlock </li></ul>
*/
private boolean canReadArchiveOrReserveWork(long absIdx) {
return archiver != null && archiver.checkCanReadArchiveOrReserveWorkSegment(absIdx);
[2/9] ignite git commit: IGNITE-7007: Decision tree code cleanup
Posted by sb...@apache.org.
IGNITE-7007: Decision tree code cleanup
This closes #3084
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a29fe352
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a29fe352
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a29fe352
Branch: refs/heads/ignite-zk
Commit: a29fe352de4fa3f66f471a4315fff097fe06c786
Parents: 3979e6a
Author: artemmalykh <am...@gridgain.com>
Authored: Fri Dec 1 20:54:59 2017 +0300
Committer: Yury Babak <yb...@gridgain.com>
Committed: Fri Dec 1 20:54:59 2017 +0300
----------------------------------------------------------------------
.../ignite/ml/math/distributed/CacheUtils.java | 2 --
.../columnbased/ColumnDecisionTreeTrainer.java | 33 +++++++++++++-------
.../org/apache/ignite/ml/util/MnistUtils.java | 17 +++++-----
.../java/org/apache/ignite/ml/util/Utils.java | 6 ++--
.../ml/trees/ColumnDecisionTreeTrainerTest.java | 3 +-
.../ColumnDecisionTreeTrainerBenchmark.java | 31 +++++++++---------
6 files changed, 51 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/main/java/org/apache/ignite/ml/math/distributed/CacheUtils.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/math/distributed/CacheUtils.java b/modules/ml/src/main/java/org/apache/ignite/ml/math/distributed/CacheUtils.java
index 6baa865..9ca167c 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/math/distributed/CacheUtils.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/math/distributed/CacheUtils.java
@@ -484,9 +484,7 @@ public class CacheUtils {
m.put(k, v);
}
- long before = System.currentTimeMillis();
cache.putAll(m);
- System.out.println("PutAll took: " + (System.currentTimeMillis() - before));
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainer.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainer.java
index 32e33f3..fec0a83 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainer.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainer.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -37,12 +38,12 @@ import java.util.stream.Stream;
import javax.cache.Cache;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteLogger;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.processors.cache.CacheEntryImpl;
-import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.ml.Trainer;
import org.apache.ignite.ml.math.Vector;
@@ -115,6 +116,9 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
/** Ignite instance. */
private final Ignite ignite;
+ /** Logger */
+ private final IgniteLogger log;
+
/**
* Construct {@link ColumnDecisionTreeTrainer}.
*
@@ -135,6 +139,7 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
this.categoricalCalculatorProvider = categoricalCalculatorProvider;
this.regCalc = regCalc;
this.ignite = ignite;
+ this.log = ignite.log();
}
/**
@@ -329,7 +334,8 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
regsCnt++;
- X.println(">>> Globally best: " + best.info + " idx time: " + findBestRegIdx + ", calculate best: " + findBestSplit + " fi: " + best.featureIdx + ", regs: " + regsCnt);
+ if (log.isDebugEnabled())
+ log.debug("Globally best: " + best.info + " idx time: " + findBestRegIdx + ", calculate best: " + findBestSplit + " fi: " + best.featureIdx + ", regs: " + regsCnt);
// Request bitset for split region.
int ind = best.info.regionIndex();
@@ -361,8 +367,10 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
if (d > curDepth) {
curDepth = d;
- X.println(">>> Depth: " + curDepth);
- X.println(">>> Cache size: " + prjsCache.size(CachePeekMode.PRIMARY));
+ if (log.isDebugEnabled()) {
+ log.debug("Depth: " + curDepth);
+ log.debug("Cache size: " + prjsCache.size(CachePeekMode.PRIMARY));
+ }
}
before = System.currentTimeMillis();
@@ -415,16 +423,19 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
},
bestRegsKeys);
- X.println(">>> Update of projs cache took " + (System.currentTimeMillis() - before));
+ if (log.isDebugEnabled())
+ log.debug("Update of projections cache time: " + (System.currentTimeMillis() - before));
before = System.currentTimeMillis();
updateSplitCache(ind, rc, featuresCnt, ig -> i -> input.affinityKey(i, ig), uuid);
- X.println(">>> Update of split cache took " + (System.currentTimeMillis() - before));
+ if (log.isDebugEnabled())
+ log.debug("Update of split cache time: " + (System.currentTimeMillis() - before));
}
else {
- X.println(">>> Best feature index: " + bestFeatureIdx + ", best infoGain " + bestInfoGain);
+ if (log.isDebugEnabled())
+ log.debug("Best split [bestFeatureIdx=" + bestFeatureIdx + ", bestInfoGain=" + bestInfoGain + "]");
break;
}
}
@@ -541,15 +552,15 @@ public class ColumnDecisionTreeTrainer<D extends ContinuousRegionInfo> implement
double[] values = ctx.values(fIdx, ign);
double[] labels = ctx.labels();
- IgniteBiTuple<Integer, Double> max = toCompare.entrySet().stream().
+ Optional<IgniteBiTuple<Integer, Double>> max = toCompare.entrySet().stream().
map(ent -> {
SplitInfo bestSplit = ctx.featureProcessor(fIdx).findBestSplit(ent.getValue(), values, labels, ent.getKey());
return new IgniteBiTuple<>(ent.getKey(), bestSplit != null ? bestSplit.infoGain() : Double.NEGATIVE_INFINITY);
}).
- max(Comparator.comparingDouble(IgniteBiTuple::get2)).
- get();
+ max(Comparator.comparingDouble(IgniteBiTuple::get2));
- return Stream.of(new CacheEntryImpl<>(e.getKey(), max));
+ return max.<Stream<Cache.Entry<SplitKey, IgniteBiTuple<Integer, Double>>>>
+ map(objects -> Stream.of(new CacheEntryImpl<>(e.getKey(), objects))).orElseGet(Stream::empty);
},
() -> IntStream.range(0, featuresCnt).mapToObj(fIdx -> SplitCache.key(fIdx, affinity.apply(ignite).apply(fIdx), trainingUUID)).collect(Collectors.toSet())
);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/main/java/org/apache/ignite/ml/util/MnistUtils.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/util/MnistUtils.java b/modules/ml/src/main/java/org/apache/ignite/ml/util/MnistUtils.java
index d69781e..a3f1d21 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/util/MnistUtils.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/util/MnistUtils.java
@@ -25,6 +25,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.ml.math.impls.vector.DenseLocalOnHeapVector;
/**
@@ -40,14 +41,14 @@ public class MnistUtils {
* @param rnd Random numbers generatror.
* @param cnt Count of samples to read.
* @return Stream of MNIST samples.
- * @throws IOException
+ * @throws IOException In case of exception.
*/
public static Stream<DenseLocalOnHeapVector> mnist(String imagesPath, String labelsPath, Random rnd, int cnt)
throws IOException {
FileInputStream isImages = new FileInputStream(imagesPath);
FileInputStream isLabels = new FileInputStream(labelsPath);
- int magic = read4Bytes(isImages); // Skip magic number.
+ read4Bytes(isImages); // Skip magic number.
int numOfImages = read4Bytes(isImages);
int imgHeight = read4Bytes(isImages);
int imgWidth = read4Bytes(isImages);
@@ -57,10 +58,6 @@ public class MnistUtils {
int numOfPixels = imgHeight * imgWidth;
- System.out.println("Magic: " + magic);
- System.out.println("Num of images: " + numOfImages);
- System.out.println("Num of pixels: " + numOfPixels);
-
double[][] vecs = new double[numOfImages][numOfPixels + 1];
for (int imgNum = 0; imgNum < numOfImages; imgNum++) {
@@ -88,7 +85,7 @@ public class MnistUtils {
* @param outPath Path to output path.
* @param rnd Random numbers generator.
* @param cnt Count of samples to read.
- * @throws IOException
+ * @throws IOException In case of exception.
*/
public static void asLIBSVM(String imagesPath, String labelsPath, String outPath, Random rnd, int cnt)
throws IOException {
@@ -109,7 +106,7 @@ public class MnistUtils {
}
catch (IOException e) {
- e.printStackTrace();
+ throw new IgniteException("Error while converting to LIBSVM.");
}
});
}
@@ -119,9 +116,9 @@ public class MnistUtils {
* Utility method for reading 4 bytes from input stream.
*
* @param is Input stream.
- * @throws IOException
+ * @throws IOException In case of exception.
*/
private static int read4Bytes(FileInputStream is) throws IOException {
return (is.read() << 24) | (is.read() << 16) | (is.read() << 8) | (is.read());
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/main/java/org/apache/ignite/ml/util/Utils.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/util/Utils.java b/modules/ml/src/main/java/org/apache/ignite/ml/util/Utils.java
index bb779e3..847b1f1 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/util/Utils.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/util/Utils.java
@@ -22,6 +22,7 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import org.apache.ignite.IgniteException;
/**
* Class with various utility methods.
@@ -34,8 +35,9 @@ public class Utils {
* @param <T> Class of original object;
* @return Deep copy of original object.
*/
+ @SuppressWarnings({"unchecked"})
public static <T> T copy(T orig) {
- Object obj = null;
+ Object obj;
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
@@ -50,7 +52,7 @@ public class Utils {
obj = in.readObject();
}
catch (IOException | ClassNotFoundException e) {
- e.printStackTrace();
+ throw new IgniteException("Couldn't copy the object.");
}
return (T)obj;
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/test/java/org/apache/ignite/ml/trees/ColumnDecisionTreeTrainerTest.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/test/java/org/apache/ignite/ml/trees/ColumnDecisionTreeTrainerTest.java b/modules/ml/src/test/java/org/apache/ignite/ml/trees/ColumnDecisionTreeTrainerTest.java
index 2b03b47..929ded9 100644
--- a/modules/ml/src/test/java/org/apache/ignite/ml/trees/ColumnDecisionTreeTrainerTest.java
+++ b/modules/ml/src/test/java/org/apache/ignite/ml/trees/ColumnDecisionTreeTrainerTest.java
@@ -26,6 +26,7 @@ import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.DoubleStream;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.ml.math.StorageConstants;
import org.apache.ignite.ml.math.Tracer;
@@ -183,7 +184,7 @@ public class ColumnDecisionTreeTrainerTest extends BaseDecisionTreeTest {
byRegion.keySet().forEach(k -> {
LabeledVectorDouble sp = byRegion.get(k).get(0);
Tracer.showAscii(sp.vector());
- System.out.println("Act: " + sp.label() + " " + " pred: " + mdl.predict(sp.vector()));
+ X.println("Actual and predicted vectors [act=" + sp.label() + " " + ", pred=" + mdl.predict(sp.vector()) + "]");
assert mdl.predict(sp.vector()) == sp.doubleLabel();
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a29fe352/modules/ml/src/test/java/org/apache/ignite/ml/trees/performance/ColumnDecisionTreeTrainerBenchmark.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/test/java/org/apache/ignite/ml/trees/performance/ColumnDecisionTreeTrainerBenchmark.java b/modules/ml/src/test/java/org/apache/ignite/ml/trees/performance/ColumnDecisionTreeTrainerBenchmark.java
index 4e7cc24..7ca5d38 100644
--- a/modules/ml/src/test/java/org/apache/ignite/ml/trees/performance/ColumnDecisionTreeTrainerBenchmark.java
+++ b/modules/ml/src/test/java/org/apache/ignite/ml/trees/performance/ColumnDecisionTreeTrainerBenchmark.java
@@ -45,6 +45,7 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.ml.Model;
import org.apache.ignite.ml.estimators.Estimators;
@@ -163,14 +164,14 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
ColumnDecisionTreeTrainer<GiniSplitCalculator.GiniData> trainer =
new ColumnDecisionTreeTrainer<>(10, ContinuousSplitCalculators.GINI.apply(ignite), RegionCalculators.GINI, RegionCalculators.MOST_COMMON, ignite);
- System.out.println(">>> Training started");
+ X.println("Training started.");
long before = System.currentTimeMillis();
DecisionTreeModel mdl = trainer.train(new BiIndexedCacheColumnDecisionTreeTrainerInput(cache, new HashMap<>(), ptsCnt, featCnt));
- System.out.println(">>> Training finished in " + (System.currentTimeMillis() - before));
+ X.println("Training finished in " + (System.currentTimeMillis() - before));
IgniteTriFunction<Model<Vector, Double>, Stream<IgniteBiTuple<Vector, Double>>, Function<Double, Double>, Double> mse = Estimators.errorsPercentage();
Double accuracy = mse.apply(mdl, testMnistStream.map(v -> new IgniteBiTuple<>(v.viewPart(0, featCnt), v.getX(featCnt))), Function.identity());
- System.out.println(">>> Errs percentage: " + accuracy);
+ X.println("Errors percentage: " + accuracy);
Assert.assertEquals(0, SplitCache.getOrCreate(ignite).size());
Assert.assertEquals(0, FeaturesCache.getOrCreate(ignite).size());
@@ -204,14 +205,14 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
ColumnDecisionTreeTrainer<GiniSplitCalculator.GiniData> trainer =
new ColumnDecisionTreeTrainer<>(10, ContinuousSplitCalculators.GINI.apply(ignite), RegionCalculators.GINI, RegionCalculators.MOST_COMMON, ignite);
- System.out.println(">>> Training started");
+ X.println("Training started");
long before = System.currentTimeMillis();
DecisionTreeModel mdl = trainer.train(new MatrixColumnDecisionTreeTrainerInput(m, new HashMap<>()));
- System.out.println(">>> Training finished in " + (System.currentTimeMillis() - before));
+ X.println("Training finished in " + (System.currentTimeMillis() - before));
IgniteTriFunction<Model<Vector, Double>, Stream<IgniteBiTuple<Vector, Double>>, Function<Double, Double>, Double> mse = Estimators.errorsPercentage();
Double accuracy = mse.apply(mdl, testMnistStream.map(v -> new IgniteBiTuple<>(v.viewPart(0, featCnt), v.getX(featCnt))), Function.identity());
- System.out.println(">>> Errs percentage: " + accuracy);
+ X.println("Errors percentage: " + accuracy);
Assert.assertEquals(0, SplitCache.getOrCreate(ignite).size());
Assert.assertEquals(0, FeaturesCache.getOrCreate(ignite).size());
@@ -252,10 +253,10 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
SparseDistributedMatrixStorage sto = (SparseDistributedMatrixStorage)m.getStorage();
long before = System.currentTimeMillis();
- System.out.println(">>> Batch loading started...");
+ X.println("Batch loading started...");
loadVectorsIntoSparseDistributedMatrixCache(sto.cache().getName(), sto.getUUID(), gen.
points(ptsPerReg, (i, rn) -> i).map(IgniteBiTuple::get2).iterator(), featCnt + 1);
- System.out.println(">>> Batch loading took " + (System.currentTimeMillis() - before) + " ms.");
+ X.println("Batch loading took " + (System.currentTimeMillis() - before) + " ms.");
for (IgniteBiTuple<Integer, DenseLocalOnHeapVector> bt : lst) {
byRegion.putIfAbsent(bt.get1(), new LinkedList<>());
@@ -268,12 +269,12 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
before = System.currentTimeMillis();
DecisionTreeModel mdl = trainer.train(new MatrixColumnDecisionTreeTrainerInput(m, catsInfo));
- System.out.println(">>> Took time(ms): " + (System.currentTimeMillis() - before));
+ X.println("Training took: " + (System.currentTimeMillis() - before) + " ms.");
byRegion.keySet().forEach(k -> {
LabeledVectorDouble sp = byRegion.get(k).get(0);
Tracer.showAscii(sp.vector());
- System.out.println("Prediction: " + mdl.predict(sp.vector()) + "label: " + sp.doubleLabel());
+ X.println("Predicted value and label [pred=" + mdl.predict(sp.vector()) + ", label=" + sp.doubleLabel() + "]");
assert mdl.predict(sp.vector()) == sp.doubleLabel();
});
}
@@ -307,16 +308,16 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
ColumnDecisionTreeTrainer<VarianceSplitCalculator.VarianceData> trainer =
new ColumnDecisionTreeTrainer<>(10, ContinuousSplitCalculators.VARIANCE, RegionCalculators.VARIANCE, regCalc, ignite);
- System.out.println(">>> Training started");
+ X.println("Training started.");
long before = System.currentTimeMillis();
DecisionTreeModel mdl = trainer.train(new MatrixColumnDecisionTreeTrainerInput(m, new HashMap<>()));
- System.out.println(">>> Training finished in " + (System.currentTimeMillis() - before));
+ X.println("Training finished in: " + (System.currentTimeMillis() - before) + " ms.");
Vector[] testVectors = vecsFromRanges(ranges, featCnt, defRng, new Random(123L), 20, f1);
IgniteTriFunction<Model<Vector, Double>, Stream<IgniteBiTuple<Vector, Double>>, Function<Double, Double>, Double> mse = Estimators.MSE();
Double accuracy = mse.apply(mdl, Arrays.stream(testVectors).map(v -> new IgniteBiTuple<>(v.viewPart(0, featCnt), v.getX(featCnt))), Function.identity());
- System.out.println(">>> MSE: " + accuracy);
+ X.println("MSE: " + accuracy);
}
/**
@@ -358,7 +359,7 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
for (int i = 0; i < vectorSize; i++)
batch.get(i).put(sampleIdx, next.getX(i));
- System.out.println(sampleIdx);
+ X.println("Sample index: " + sampleIdx);
if (sampleIdx % batchSize == 0) {
batch.keySet().forEach(fi -> streamer.addData(new SparseMatrixKey(fi, uuid, fi), batch.get(fi)));
IntStream.range(0, vectorSize).forEach(i -> batch.put(i, new HashMap<>()));
@@ -396,7 +397,7 @@ public class ColumnDecisionTreeTrainerBenchmark extends BaseDecisionTreeTest {
sampleIdx++;
if (sampleIdx % 1000 == 0)
- System.out.println(">>> Loaded " + sampleIdx + " vectors.");
+ System.out.println("Loaded: " + sampleIdx + " vectors.");
}
}
}
[4/9] ignite git commit: IGNITE-7096: Missed dependencies in examples
pom files
Posted by sb...@apache.org.
IGNITE-7096: Missed dependencies in examples pom files
this closes #3132
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/93988132
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/93988132
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/93988132
Branch: refs/heads/ignite-zk
Commit: 93988132c7850fc4f443d223490a41d7aa94ad9f
Parents: 104a7c8
Author: YuriBabak <y....@gmail.com>
Authored: Sat Dec 2 17:55:12 2017 +0300
Committer: YuriBabak <y....@gmail.com>
Committed: Sat Dec 2 17:55:12 2017 +0300
----------------------------------------------------------------------
examples/pom-standalone-lgpl.xml | 5 +++++
examples/pom-standalone.xml | 5 +++++
2 files changed, 10 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/93988132/examples/pom-standalone-lgpl.xml
----------------------------------------------------------------------
diff --git a/examples/pom-standalone-lgpl.xml b/examples/pom-standalone-lgpl.xml
index 3577d9c..2f6ff9c 100644
--- a/examples/pom-standalone-lgpl.xml
+++ b/examples/pom-standalone-lgpl.xml
@@ -168,6 +168,11 @@
<artifactId>ignite-ml</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ <version>1.2</version>
+ </dependency>
</dependencies>
</profile>
http://git-wip-us.apache.org/repos/asf/ignite/blob/93988132/examples/pom-standalone.xml
----------------------------------------------------------------------
diff --git a/examples/pom-standalone.xml b/examples/pom-standalone.xml
index 2c7cd1b..bf2117a 100644
--- a/examples/pom-standalone.xml
+++ b/examples/pom-standalone.xml
@@ -168,6 +168,11 @@
<artifactId>ignite-ml</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ <version>1.2</version>
+ </dependency>
</dependencies>
</profile>
[9/9] ignite git commit: zk
Posted by sb...@apache.org.
zk
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/74330bc1
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/74330bc1
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/74330bc1
Branch: refs/heads/ignite-zk
Commit: 74330bc142aa3f5d7bf18a4eb36f50e3b776f8dd
Parents: 3259573
Author: sboikov <sb...@gridgain.com>
Authored: Mon Dec 4 10:51:16 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Dec 4 10:51:16 2017 +0300
----------------------------------------------------------------------
.../spi/discovery/zk/internal/ZkInternalFailNodeMessage.java | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/74330bc1/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkInternalFailNodeMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkInternalFailNodeMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkInternalFailNodeMessage.java
index b289af1..a97289d 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkInternalFailNodeMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkInternalFailNodeMessage.java
@@ -26,6 +26,9 @@ import org.jetbrains.annotations.Nullable;
*/
public class ZkInternalFailNodeMessage implements ZkInternalMessage {
/** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
final UUID nodeId;
/** */
[3/9] ignite git commit: Ignite-7094: Javadoc warnings in ml module
Posted by sb...@apache.org.
Ignite-7094: Javadoc warnings in ml module
this closes #3131
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/104a7c8f
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/104a7c8f
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/104a7c8f
Branch: refs/heads/ignite-zk
Commit: 104a7c8fa36bbfb867476f3e26cd37c89f242281
Parents: a29fe35
Author: YuriBabak <y....@gmail.com>
Authored: Sat Dec 2 17:35:32 2017 +0300
Committer: YuriBabak <y....@gmail.com>
Committed: Sat Dec 2 17:35:32 2017 +0300
----------------------------------------------------------------------
.../impls/storage/vector/SparseDistributedVectorStorage.java | 3 ++-
.../org/apache/ignite/ml/trees/ContinuousSplitCalculator.java | 3 ++-
.../columnbased/CacheColumnDecisionTreeTrainerInput.java | 7 +++----
.../trainers/columnbased/ColumnDecisionTreeTrainerInput.java | 2 +-
.../columnbased/MatrixColumnDecisionTreeTrainerInput.java | 5 +++--
.../columnbased/vectors/CategoricalFeatureProcessor.java | 3 ++-
.../trees/trainers/columnbased/vectors/FeatureProcessor.java | 5 +++--
7 files changed, 16 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/math/impls/storage/vector/SparseDistributedVectorStorage.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/math/impls/storage/vector/SparseDistributedVectorStorage.java b/modules/ml/src/main/java/org/apache/ignite/ml/math/impls/storage/vector/SparseDistributedVectorStorage.java
index af1c2cd..0496d05 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/math/impls/storage/vector/SparseDistributedVectorStorage.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/math/impls/storage/vector/SparseDistributedVectorStorage.java
@@ -36,9 +36,10 @@ import org.apache.ignite.ml.math.distributed.CacheUtils;
import org.apache.ignite.ml.math.distributed.DistributedStorage;
import org.apache.ignite.ml.math.distributed.keys.RowColMatrixKey;
import org.apache.ignite.ml.math.distributed.keys.impl.SparseMatrixKey;
+import org.apache.ignite.ml.math.impls.vector.SparseDistributedVector;
/**
- * {@link VectorStorage} implementation for {@link /*SparseDistributedVector}.
+ * {@link VectorStorage} implementation for {@link SparseDistributedVector}.
*/
public class SparseDistributedVectorStorage extends CacheUtils implements VectorStorage, StorageConstants, DistributedStorage<RowColMatrixKey> {
/** Cache name used for all instances of {@link SparseDistributedVectorStorage}. */
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/ContinuousSplitCalculator.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/ContinuousSplitCalculator.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/ContinuousSplitCalculator.java
index f9b81d0..3a0e9da 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/ContinuousSplitCalculator.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/ContinuousSplitCalculator.java
@@ -19,6 +19,7 @@ package org.apache.ignite.ml.trees;
import java.util.stream.DoubleStream;
import org.apache.ignite.ml.trees.trainers.columnbased.vectors.SplitInfo;
+import org.apache.ignite.ml.trees.trainers.columnbased.vectors.ContinuousFeatureProcessor;
/**
* This class is used for calculation of best split by continuous feature.
@@ -31,7 +32,7 @@ public interface ContinuousSplitCalculator<C extends ContinuousRegionInfo> {
*
* @param s Stream of labels in this region.
* @param l Index of sample projection on this feature in array sorted by this projection value and intervals
- * bitsets. ({@see org.apache.ignite.ml.trees.trainers.columnbased.vectors.ContinuousFeatureProcessor}).
+ * bitsets. ({@link ContinuousFeatureProcessor}).
* @return Region info.
*/
C calculateRegionInfo(DoubleStream s, int l);
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/CacheColumnDecisionTreeTrainerInput.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/CacheColumnDecisionTreeTrainerInput.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/CacheColumnDecisionTreeTrainerInput.java
index 9518caf..40927b7 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/CacheColumnDecisionTreeTrainerInput.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/CacheColumnDecisionTreeTrainerInput.java
@@ -31,7 +31,7 @@ import org.apache.ignite.ml.math.functions.IgniteFunction;
import org.apache.ignite.ml.math.functions.IgniteSupplier;
/**
- * Adapter of a given cache to {@see CacheColumnDecisionTreeTrainerInput}
+ * Adapter of a given cache to {@link CacheColumnDecisionTreeTrainerInput}
*
* @param <K> Class of keys of the cache.
* @param <V> Class of values of the cache.
@@ -66,11 +66,10 @@ public abstract class CacheColumnDecisionTreeTrainerInput<K, V> implements Colum
private final IgniteFunction<V, DoubleStream> labelsMapper;
/**
- * Constructs input for {@see org.apache.ignite.ml.trees.trainers.columnbased.ColumnDecisionTreeTrainer}.
+ * Constructs input for {@link ColumnDecisionTreeTrainer}.
*
* @param c Cache.
- * @param valuesMapper Function for mapping cache entry to stream used by {@link
- * org.apache.ignite.ml.trees.trainers.columnbased.ColumnDecisionTreeTrainer}.
+ * @param valuesMapper Function for mapping cache entry to stream used by {@link ColumnDecisionTreeTrainer}.
* @param labelsMapper Function used for mapping cache value to labels array.
* @param keyMapper Function used for mapping feature index to the cache key.
* @param catFeaturesInfo Information about which features are categorical in form of feature index -> number of
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainerInput.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainerInput.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainerInput.java
index 94331f7..bf8790b 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainerInput.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/ColumnDecisionTreeTrainerInput.java
@@ -23,7 +23,7 @@ import org.apache.ignite.Ignite;
import org.apache.ignite.lang.IgniteBiTuple;
/**
- * Input for {@see ColumnDecisionTreeTrainer}.
+ * Input for {@link ColumnDecisionTreeTrainer}.
*/
public interface ColumnDecisionTreeTrainerInput {
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/MatrixColumnDecisionTreeTrainerInput.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/MatrixColumnDecisionTreeTrainerInput.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/MatrixColumnDecisionTreeTrainerInput.java
index 9a11902..3da6bad 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/MatrixColumnDecisionTreeTrainerInput.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/MatrixColumnDecisionTreeTrainerInput.java
@@ -30,16 +30,17 @@ import org.apache.ignite.ml.math.distributed.keys.impl.SparseMatrixKey;
import org.apache.ignite.ml.math.functions.IgniteFunction;
import org.apache.ignite.ml.math.impls.matrix.SparseDistributedMatrix;
import org.apache.ignite.ml.math.impls.storage.matrix.SparseDistributedMatrixStorage;
+import org.apache.ignite.ml.math.StorageConstants;
import org.jetbrains.annotations.NotNull;
/**
* Adapter of SparseDistributedMatrix to ColumnDecisionTreeTrainerInput.
- * Sparse SparseDistributedMatrix should be in {@see org.apache.ignite.ml.math.StorageConstants#COLUMN_STORAGE_MODE} and
+ * Sparse SparseDistributedMatrix should be in {@link StorageConstants#COLUMN_STORAGE_MODE} and
* should contain samples in rows last position in row being label of this sample.
*/
public class MatrixColumnDecisionTreeTrainerInput extends CacheColumnDecisionTreeTrainerInput<RowColMatrixKey, Map<Integer, Double>> {
/**
- * @param m Sparse SparseDistributedMatrix should be in {@see org.apache.ignite.ml.math.StorageConstants#COLUMN_STORAGE_MODE}
+ * @param m Sparse SparseDistributedMatrix should be in {@link StorageConstants#COLUMN_STORAGE_MODE}
* containing samples in rows last position in row being label of this sample.
* @param catFeaturesInfo Information about which features are categorical in form of feature index -> number of
* categories.
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/CategoricalFeatureProcessor.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/CategoricalFeatureProcessor.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/CategoricalFeatureProcessor.java
index 9469768..3232ac2 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/CategoricalFeatureProcessor.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/CategoricalFeatureProcessor.java
@@ -34,12 +34,13 @@ import org.apache.ignite.ml.math.functions.IgniteFunction;
import org.apache.ignite.ml.trees.CategoricalRegionInfo;
import org.apache.ignite.ml.trees.CategoricalSplitInfo;
import org.apache.ignite.ml.trees.RegionInfo;
+import org.apache.ignite.ml.trees.trainers.columnbased.ColumnDecisionTreeTrainer;
import org.apache.ignite.ml.trees.trainers.columnbased.RegionProjection;
import static org.apache.ignite.ml.trees.trainers.columnbased.vectors.FeatureVectorProcessorUtils.splitByBitSet;
/**
- * Categorical feature vector processor implementation used by {@see ColumnDecisionTreeTrainer}.
+ * Categorical feature vector processor implementation used by {@link ColumnDecisionTreeTrainer}.
*/
public class CategoricalFeatureProcessor
implements FeatureProcessor<CategoricalRegionInfo, CategoricalSplitInfo<CategoricalRegionInfo>> {
http://git-wip-us.apache.org/repos/asf/ignite/blob/104a7c8f/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/FeatureProcessor.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/FeatureProcessor.java b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/FeatureProcessor.java
index cb8f5c2..56508e5 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/FeatureProcessor.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/trees/trainers/columnbased/vectors/FeatureProcessor.java
@@ -20,10 +20,11 @@ package org.apache.ignite.ml.trees.trainers.columnbased.vectors;
import com.zaxxer.sparsebits.SparseBitSet;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.ml.trees.RegionInfo;
+import org.apache.ignite.ml.trees.trainers.columnbased.ColumnDecisionTreeTrainer;
import org.apache.ignite.ml.trees.trainers.columnbased.RegionProjection;
/**
- * Base interface for feature processors used in {@see org.apache.ignite.ml.trees.trainers.columnbased.ColumnDecisionTreeTrainer}
+ * Base interface for feature processors used in {@link ColumnDecisionTreeTrainer}
*
* @param <D> Class representing data of regions resulted from split.
* @param <S> Class representing data of split.
@@ -68,7 +69,7 @@ public interface FeatureProcessor<D extends RegionInfo, S extends SplitInfo<D>>
/**
* Splits given region using bitset which maps data point to left or right subregion. This method is used iff the
* optimal split is received on feature of different type, therefore information about regions is limited to the
- * {@see RegionInfo} class which is base for all classes used to represent region data.
+ * {@link RegionInfo} class which is base for all classes used to represent region data.
*
* @param bs Bitset which maps data point to left or right subregion.
* @param leftData Data of the left subregion.
[5/9] ignite git commit: ignite-7049 Fixed error in tx timeout
processing for optimistic/serializable tx
Posted by sb...@apache.org.
ignite-7049 Fixed error in tx timeout processing for optimistic/serializable tx
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/cd0d2ebf
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/cd0d2ebf
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/cd0d2ebf
Branch: refs/heads/ignite-zk
Commit: cd0d2ebf6a4294be851fb316f9b2f6b4ce6cb321
Parents: 9398813
Author: Aleksei Scherbakov <al...@gmail.com>
Authored: Sat Dec 2 20:36:29 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Sat Dec 2 20:36:29 2017 +0300
----------------------------------------------------------------------
...arOptimisticSerializableTxPrepareFuture.java | 2 +-
.../transactions/TxRollbackOnTimeoutTest.java | 186 ++++++++++++++++++-
2 files changed, 186 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd0d2ebf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
index 1da0589..beb1e16 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearOptimisticSerializableTxPrepareFuture.java
@@ -184,7 +184,7 @@ public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptim
}
}
- if (e instanceof IgniteTxOptimisticCheckedException || e instanceof IgniteTxTimeoutCheckedException) {
+ if (e instanceof IgniteTxOptimisticCheckedException) {
if (m != null)
tx.removeMapping(m.primary().id());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/cd0d2ebf/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
index e1c6c10..6aa3bdd 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutTest.java
@@ -18,24 +18,31 @@
package org.apache.ignite.internal.processors.cache.transactions;
import java.util.Collection;
+import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.cache.CacheException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.testframework.GridTestUtils;
@@ -44,8 +51,11 @@ import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionDeadlockException;
import org.apache.ignite.transactions.TransactionIsolation;
+import org.apache.ignite.transactions.TransactionOptimisticException;
import org.apache.ignite.transactions.TransactionTimeoutException;
+import org.jsr166.LongAdder8;
+import static java.lang.Thread.sleep;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
@@ -56,6 +66,9 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
*/
public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
/** */
+ private static final long DURATION = 60 * 1000L;
+
+ /** */
private static final long TX_MIN_TIMEOUT = 1;
/** */
@@ -73,6 +86,8 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(IP_FINDER);
+ cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+
boolean client = "client".equals(igniteInstanceName);
cfg.setClientMode(client);
@@ -373,6 +388,175 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
}
/**
+ * Test timeouts with random values and different tx configurations.
+ */
+ public void testRandomMixedTxConfigurations() throws Exception {
+ final Ignite client = startClient();
+
+ final AtomicBoolean stop = new AtomicBoolean();
+
+ final long seed = System.currentTimeMillis();
+
+ final Random r = new Random(seed);
+
+ log.info("Using seed: " + seed);
+
+ final int threadsCnt = Runtime.getRuntime().availableProcessors() * 2;
+
+ for (int k = 0; k < threadsCnt; k++)
+ grid(0).cache(CACHE_NAME).put(k, (long)0);
+
+ final TransactionConcurrency[] TC_VALS = TransactionConcurrency.values();
+ final TransactionIsolation[] TI_VALS = TransactionIsolation.values();
+
+ final LongAdder8 cntr0 = new LongAdder8();
+ final LongAdder8 cntr1 = new LongAdder8();
+ final LongAdder8 cntr2 = new LongAdder8();
+ final LongAdder8 cntr3 = new LongAdder8();
+
+ final IgniteInternalFuture<?> fut = multithreadedAsync(new Runnable() {
+ @Override public void run() {
+ while (!stop.get()) {
+ int nodeId = r.nextInt(GRID_CNT + 1);
+
+ Ignite node = nodeId == GRID_CNT || nearCacheEnabled() ? client : grid(nodeId);
+
+ TransactionConcurrency conc = TC_VALS[r.nextInt(TC_VALS.length)];
+ TransactionIsolation isolation = TI_VALS[r.nextInt(TI_VALS.length)];
+
+ int k = r.nextInt(threadsCnt);
+
+ long timeout = r.nextInt(200) + 50;
+
+ // Roughly 50% of transactions should time out.
+ try (Transaction tx = node.transactions().txStart(conc, isolation, timeout, 1)) {
+ cntr0.add(1);
+
+ final Long v = (Long)node.cache(CACHE_NAME).get(k);
+
+ final int delay = r.nextInt(400);
+
+ if (delay > 0)
+ sleep(delay);
+
+ node.cache(CACHE_NAME).put(k, v + 1);
+
+ tx.commit();
+
+ cntr1.add(1);
+ }
+ catch (TransactionOptimisticException | InterruptedException e) {
+ // Expected.
+ cntr3.add(1);
+ }
+ catch (TransactionTimeoutException e) {
+ cntr2.add(1);
+ }
+ catch (CacheException e) {
+ assertEquals(TransactionTimeoutException.class, X.getCause(e).getClass());
+
+ cntr2.add(1);
+ }
+ }
+ }
+ }, threadsCnt, "tx-async-thread");
+
+ sleep(DURATION);
+
+ stop.set(true);
+
+ fut.get(10_000);
+
+ log.info("Tx test stats: started=" + cntr0.sum() +
+ ", completed=" + cntr1.sum() +
+ ", failed=" + cntr3.sum() +
+ ", timedOut=" + cntr2.sum());
+
+ assertEquals("Expected finished count same as started count", cntr0.sum(), cntr1.sum() + cntr2.sum() + cntr3.sum());
+ }
+
+ /**
+ * Tests timeout on DHT primary node for all tx configurations.
+ *
+ * @throws Exception If failed.
+ */
+ public void testTimeoutOnPrimaryDHTNode() throws Exception {
+ final ClusterNode n0 = grid(0).affinity(CACHE_NAME).mapKeyToNode(0);
+
+ final Ignite prim = G.ignite(n0.id());
+
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ for (TransactionIsolation isolation : TransactionIsolation.values())
+ testTimeoutOnPrimaryDhtNode0(prim, concurrency, isolation);
+ }
+ }
+
+ /**
+ *
+ * @param prim Primary node.
+ * @param conc Concurrency.
+ * @param isolation Isolation.
+
+ * @throws Exception If failed.
+ */
+ private void testTimeoutOnPrimaryDhtNode0(final Ignite prim, final TransactionConcurrency conc,
+ final TransactionIsolation isolation)
+ throws Exception {
+
+ log.info("concurrency=" + conc + ", isolation=" + isolation);
+
+ // Force timeout on primary DHT node by blocking DHT prepare response.
+ toggleBlocking(GridDhtTxPrepareResponse.class, prim, true);
+
+ final int val = 0;
+
+ try {
+ multithreaded(new Runnable() {
+ @Override public void run() {
+ try (Transaction txOpt = prim.transactions().txStart(conc, isolation, 300, 1)) {
+
+ prim.cache(CACHE_NAME).put(val, val);
+
+ txOpt.commit();
+ }
+ }
+ }, 1, "tx-async-thread");
+
+ fail();
+ }
+ catch (TransactionTimeoutException e) {
+ // Expected.
+ }
+
+ toggleBlocking(GridDhtTxPrepareResponse.class, prim, false);
+
+ AffinityTopologyVersion topVer = new AffinityTopologyVersion(GRID_CNT + 1, 0);
+
+ for (Ignite ignite : G.allGrids())
+ ((IgniteEx)ignite).context().cache().context().partitionReleaseFuture(topVer).get(10_000);
+ }
+
+ /**
+ * @param cls Message class.
+ * @param nodeToBlock Node to block.
+ * @param block Block.
+ */
+ private void toggleBlocking(Class<? extends Message> cls, Ignite nodeToBlock, boolean block) {
+ for (Ignite ignite : G.allGrids()) {
+ if (ignite == nodeToBlock)
+ continue;
+
+ final TestRecordingCommunicationSpi spi =
+ (TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
+
+ if (block)
+ spi.blockMessages(cls, nodeToBlock.name());
+ else
+ spi.stopBlock(true);
+ }
+ }
+
+ /**
* @param concurrency Concurrency.
* @param isolation Isolation.
* @param op Operation to test.
@@ -652,4 +836,4 @@ public class TxRollbackOnTimeoutTest extends GridCommonAbstractTest {
fut2.get();
}
-}
\ No newline at end of file
+}
[8/9] ignite git commit: Merge remote-tracking branch
'remotes/origin/master' into ignite-zk
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/master' into ignite-zk
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3259573c
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3259573c
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3259573c
Branch: refs/heads/ignite-zk
Commit: 3259573c3412b28187cf5af2b0c5e801818d6a7a
Parents: 73cea16 6101fde
Author: sboikov <sb...@gridgain.com>
Authored: Mon Dec 4 10:49:12 2017 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Mon Dec 4 10:49:12 2017 +0300
----------------------------------------------------------------------
examples/pom-standalone-lgpl.xml | 5 +
examples/pom-standalone.xml | 5 +
.../ml/clustering/FuzzyCMeansExample.java | 125 +++++
.../examples/ml/clustering/package-info.java | 22 +
.../DummyPersistenceCompatibilityTest.java | 2 +-
...arOptimisticSerializableTxPrepareFuture.java | 2 +-
.../wal/FileWriteAheadLogManager.java | 27 +-
.../transactions/TxRollbackOnTimeoutTest.java | 186 ++++++-
.../ignite/ml/FuzzyCMeansModelFormat.java | 76 +++
.../ml/clustering/BaseFuzzyCMeansClusterer.java | 90 ++++
.../FuzzyCMeansDistributedClusterer.java | 512 +++++++++++++++++++
.../clustering/FuzzyCMeansLocalClusterer.java | 254 +++++++++
.../ignite/ml/clustering/FuzzyCMeansModel.java | 88 ++++
.../ignite/ml/clustering/WeightedClusterer.java | 2 +-
.../ignite/ml/math/distributed/CacheUtils.java | 2 -
.../vector/SparseDistributedVectorStorage.java | 3 +-
.../ml/trees/ContinuousSplitCalculator.java | 3 +-
.../CacheColumnDecisionTreeTrainerInput.java | 7 +-
.../columnbased/ColumnDecisionTreeTrainer.java | 33 +-
.../ColumnDecisionTreeTrainerInput.java | 2 +-
.../MatrixColumnDecisionTreeTrainerInput.java | 5 +-
.../vectors/CategoricalFeatureProcessor.java | 3 +-
.../columnbased/vectors/FeatureProcessor.java | 5 +-
.../org/apache/ignite/ml/util/MnistUtils.java | 17 +-
.../java/org/apache/ignite/ml/util/Utils.java | 6 +-
.../ml/clustering/ClusteringTestSuite.java | 4 +-
.../FuzzyCMeansDistributedClustererTest.java | 177 +++++++
.../FuzzyCMeansLocalClustererTest.java | 203 ++++++++
.../ml/trees/ColumnDecisionTreeTrainerTest.java | 3 +-
.../ColumnDecisionTreeTrainerBenchmark.java | 31 +-
.../Client/Cache/CacheTest.cs | 4 +-
.../Client/Cache/CreateCacheTest.cs | 4 +-
.../Client/Cache/ScanQueryTest.cs | 4 +-
.../Client/ClientConnectionTest.cs | 2 +-
.../Client/RawSocketTest.cs | 3 +-
.../Apache.Ignite.Core.csproj | 2 +-
.../Client/ClientStatusCode.cs | 57 +++
.../Client/IgniteClientException.cs | 24 +-
.../Impl/Client/Cache/CacheClient.cs | 8 +-
.../Impl/Client/ClientSocket.cs | 8 +-
.../Impl/Client/ClientStatus.cs | 32 --
41 files changed, 1919 insertions(+), 129 deletions(-)
----------------------------------------------------------------------
[6/9] ignite git commit: IGNITE-5246: Fuzzy c-means (FCM)
Posted by sb...@apache.org.
IGNITE-5246: Fuzzy c-means (FCM)
this closes #2969
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b13f088b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b13f088b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b13f088b
Branch: refs/heads/ignite-zk
Commit: b13f088b210b66b4b38c260b48d6853e03f46a6c
Parents: cd0d2eb
Author: Ilya Nozhkin <no...@gmail.com>
Authored: Sun Dec 3 23:08:49 2017 +0300
Committer: Yury Babak <yb...@gridgain.com>
Committed: Sun Dec 3 23:08:49 2017 +0300
----------------------------------------------------------------------
.../ml/clustering/FuzzyCMeansExample.java | 125 +++++
.../examples/ml/clustering/package-info.java | 22 +
.../ignite/ml/FuzzyCMeansModelFormat.java | 76 +++
.../ml/clustering/BaseFuzzyCMeansClusterer.java | 90 ++++
.../FuzzyCMeansDistributedClusterer.java | 512 +++++++++++++++++++
.../clustering/FuzzyCMeansLocalClusterer.java | 254 +++++++++
.../ignite/ml/clustering/FuzzyCMeansModel.java | 88 ++++
.../ignite/ml/clustering/WeightedClusterer.java | 2 +-
.../ml/clustering/ClusteringTestSuite.java | 4 +-
.../FuzzyCMeansDistributedClustererTest.java | 177 +++++++
.../FuzzyCMeansLocalClustererTest.java | 203 ++++++++
11 files changed, 1551 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/FuzzyCMeansExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/FuzzyCMeansExample.java b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/FuzzyCMeansExample.java
new file mode 100644
index 0000000..9c47186
--- /dev/null
+++ b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/FuzzyCMeansExample.java
@@ -0,0 +1,125 @@
+/*
+ * 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.ignite.examples.ml.clustering;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.ml.clustering.BaseFuzzyCMeansClusterer;
+import org.apache.ignite.ml.clustering.FuzzyCMeansDistributedClusterer;
+import org.apache.ignite.ml.clustering.FuzzyCMeansModel;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.EuclideanDistance;
+import org.apache.ignite.ml.math.StorageConstants;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.impls.matrix.SparseDistributedMatrix;
+import org.apache.ignite.thread.IgniteThread;
+
+/**
+ * This example shows how to use Fuzzy C-Means clusterer
+ * ({@link org.apache.ignite.ml.clustering.FuzzyCMeansDistributedClusterer}).
+ */
+public final class FuzzyCMeansExample {
+ /**
+ * Executes example.
+ *
+ * @param args Command line arguments, none required.
+ */
+ public static void main(String[] args) throws InterruptedException {
+ System.out.println();
+ System.out.println(">>> Fuzzy C-Means usage example started.");
+ // Start ignite grid.
+ try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+ System.out.println(">>> Ignite grid started.");
+
+ // Start new Ignite thread.
+ IgniteThread igniteThread = new IgniteThread(ignite.configuration().getIgniteInstanceName(),
+ FuzzyCMeansExample.class.getSimpleName(),
+ () -> {
+
+ // Distance measure that computes distance between two points.
+ DistanceMeasure distanceMeasure = new EuclideanDistance();
+
+ // "Fuzziness" - specific constant that is used in membership calculation (1.0+-eps ~ K-Means).
+ double exponentialWeight = 2.0;
+
+ // Condition that indicated when algorithm must stop.
+ // In this example algorithm stops if memberships have changed insignificantly.
+ BaseFuzzyCMeansClusterer.StopCondition stopCond =
+ BaseFuzzyCMeansClusterer.StopCondition.STABLE_MEMBERSHIPS;
+
+ // Maximum difference between new and old membership values with which algorithm will continue to work.
+ double maxDelta = 0.01;
+
+ // The maximum number of FCM iterations.
+ int maxIterations = 50;
+
+ // Value that is used to initialize random numbers generator. You can choose it randomly.
+ Long seed = null;
+
+ // Number of steps of primary centers selection (more steps more candidates).
+ int initializationSteps = 2;
+
+ // Number of K-Means iteration that is used to choose required number of primary centers from candidates.
+ int kMeansMaxIterations = 50;
+
+ // Create new distributed clusterer with parameters described above.
+ System.out.println(">>> Create new Distributed Fuzzy C-Means clusterer.");
+ FuzzyCMeansDistributedClusterer clusterer = new FuzzyCMeansDistributedClusterer(
+ distanceMeasure, exponentialWeight, stopCond, maxDelta, maxIterations,
+ seed, initializationSteps, kMeansMaxIterations);
+
+ // Create sample data.
+ double[][] points = new double[][]{{-10, -10}, {-9, -11}, {-10, -9}, {-11, -9},
+ {10, 10}, {9, 11}, {10, 9}, {11, 9},
+ {-10, 10}, {-9, 11}, {-10, 9}, {-11, 9},
+ {10, -10}, {9, -11}, {10, -9}, {11, -9}};
+
+ // Initialize matrix of data points. Each row contains one point.
+ int rows = points.length;
+ int cols = points[0].length;
+
+ System.out.println(">>> Create the matrix that contains sample points.");
+ SparseDistributedMatrix pntMatrix = new SparseDistributedMatrix(rows, cols,
+ StorageConstants.ROW_STORAGE_MODE, StorageConstants.RANDOM_ACCESS_MODE);
+
+ // Store points into matrix.
+ pntMatrix.assign(points);
+
+ // Call clusterization method with some number of centers.
+ // It returns model that can predict results for new points.
+ System.out.println(">>> Perform clusterization.");
+ int numCenters = 4;
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, numCenters);
+
+ // You can also get centers of clusters that is computed by Fuzzy C-Means algorithm.
+ Vector[] centers = mdl.centers();
+
+ StringBuilder results = new StringBuilder(">>> Results:\n");
+ results.append(">>> 1st center: " + centers[0].get(0) + " " + centers[0].get(1) + "\n");
+ results.append(">>> 2nd center: " + centers[1].get(0) + " " + centers[1].get(1) + "\n");
+ results.append(">>> 3rd center: " + centers[2].get(0) + " " + centers[2].get(1) + "\n");
+ results.append(">>> 4th center: " + centers[3].get(0) + " " + centers[3].get(1) + "\n");
+
+ System.out.println(results.toString());
+ });
+
+ igniteThread.start();
+ igniteThread.join();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/package-info.java b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/package-info.java
new file mode 100644
index 0000000..7051912
--- /dev/null
+++ b/examples/src/main/ml/org/apache/ignite/examples/ml/clustering/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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 description. -->
+ * Clustering examples.
+ */
+package org.apache.ignite.examples.ml.clustering;
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/FuzzyCMeansModelFormat.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/FuzzyCMeansModelFormat.java b/modules/ml/src/main/java/org/apache/ignite/ml/FuzzyCMeansModelFormat.java
new file mode 100644
index 0000000..2b27e86
--- /dev/null
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/FuzzyCMeansModelFormat.java
@@ -0,0 +1,76 @@
+/*
+ * 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.ignite.ml;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.Vector;
+
+/** Fuzzy C-Means model representation. */
+public class FuzzyCMeansModelFormat implements Serializable {
+ /** Centers of clusters. */
+ private final Vector[] centers;
+
+ /** Distance measure. */
+ private final DistanceMeasure measure;
+
+ /**
+ * Constructor that retains result of clusterization and distance measure.
+ *
+ * @param centers Centers found while clusterization.
+ * @param measure Distance measure.
+ */
+ public FuzzyCMeansModelFormat(Vector[] centers, DistanceMeasure measure) {
+ this.centers = centers;
+ this.measure = measure;
+ }
+
+ /** Distance measure used while clusterization. */
+ public DistanceMeasure getMeasure() {
+ return measure;
+ }
+
+ /** Get cluster centers. */
+ public Vector[] getCenters() {
+ return centers;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int res = 1;
+
+ res = res * 37 + measure.hashCode();
+ res = res * 37 + Arrays.hashCode(centers);
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object obj) {
+ if (this == obj)
+ return true;
+
+ if (obj == null || getClass() != obj.getClass())
+ return false;
+
+ FuzzyCMeansModelFormat that = (FuzzyCMeansModelFormat) obj;
+
+ return measure.equals(that.measure) && Arrays.deepEquals(centers, that.centers);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/BaseFuzzyCMeansClusterer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/BaseFuzzyCMeansClusterer.java b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/BaseFuzzyCMeansClusterer.java
new file mode 100644
index 0000000..65aaeee
--- /dev/null
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/BaseFuzzyCMeansClusterer.java
@@ -0,0 +1,90 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.Matrix;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.exceptions.ConvergenceException;
+import org.apache.ignite.ml.math.exceptions.MathIllegalArgumentException;
+
+/** The abstract class that defines the basic interface of Fuzzy C-Means clusterers */
+public abstract class BaseFuzzyCMeansClusterer<T extends Matrix> implements Clusterer<T, FuzzyCMeansModel> {
+ /** Distance measure. */
+ protected DistanceMeasure measure;
+
+ /** Specific constant which is used in calculating of membership matrix. */
+ protected double exponentialWeight;
+
+ /** The maximum distance between old and new centers or the maximum difference between new and old membership matrix
+ * elements for which algorithm must stop. */
+ protected double maxDelta;
+
+ /** The flag that tells when algorithm should stop. */
+ protected StopCondition stopCond;
+
+ /**
+ * Constructor that stores some required parameters.
+ *
+ * @param measure Distance measure.
+ * @param exponentialWeight Specific constant which is used in calculating of membership matrix.
+ * @param stopCond Flag that tells when algorithm should stop.
+ * @param maxDelta The maximum distance between old and new centers or maximum difference between new and old
+ * membership matrix elements for which algorithm must stop.
+ */
+ protected BaseFuzzyCMeansClusterer(DistanceMeasure measure, double exponentialWeight, StopCondition stopCond,
+ double maxDelta) {
+ this.measure = measure;
+ this.exponentialWeight = exponentialWeight;
+ this.stopCond = stopCond;
+ this.maxDelta = maxDelta;
+ }
+
+ /**
+ * Perform a cluster analysis on the given set of points.
+ *
+ * @param points The set of points.
+ * @return A list of clusters.
+ * @throws MathIllegalArgumentException If points are null or the number of data points is not compatible with this
+ * clusterer.
+ * @throws ConvergenceException If the algorithm has not yet converged after the maximum number of iterations has
+ * been exceeded.
+ */
+ public abstract FuzzyCMeansModel cluster(T points, int k);
+
+ /**
+ * Calculates the distance between two vectors. * with the configured {@link DistanceMeasure}.
+ *
+ * @return The distance between two points.
+ */
+ protected double distance(final Vector v1, final Vector v2) {
+ return measure.compute(v1, v2);
+ }
+
+ /** Enumeration that contains different conditions under which algorithm must stop. */
+ public enum StopCondition {
+ /** Algorithm stops if the maximum distance between new and old centers is less than {@link #maxDelta}. */
+ STABLE_CENTERS,
+
+ /**
+ * Algorithm stops if the maximum difference between elements of new and old membership matrix is less than
+ * {@link #maxDelta}.
+ */
+ STABLE_MEMBERSHIPS
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClusterer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClusterer.java b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClusterer.java
new file mode 100644
index 0000000..a5cd871
--- /dev/null
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClusterer.java
@@ -0,0 +1,512 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import javax.cache.Cache;
+import org.apache.ignite.internal.util.GridArgumentCheck;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.VectorUtils;
+import org.apache.ignite.ml.math.distributed.CacheUtils;
+import org.apache.ignite.ml.math.distributed.keys.impl.SparseMatrixKey;
+import org.apache.ignite.ml.math.exceptions.ConvergenceException;
+import org.apache.ignite.ml.math.exceptions.MathIllegalArgumentException;
+import org.apache.ignite.ml.math.functions.Functions;
+import org.apache.ignite.ml.math.functions.IgniteBiFunction;
+import org.apache.ignite.ml.math.functions.IgniteSupplier;
+import org.apache.ignite.ml.math.impls.matrix.DenseLocalOnHeapMatrix;
+import org.apache.ignite.ml.math.impls.matrix.SparseDistributedMatrix;
+import org.apache.ignite.ml.math.impls.storage.matrix.SparseDistributedMatrixStorage;
+import org.apache.ignite.ml.math.impls.vector.DenseLocalOnHeapVector;
+import org.apache.ignite.ml.math.util.MatrixUtil;
+
+/** This class implements distributed version of Fuzzy C-Means clusterization of equal-weighted points. */
+public class FuzzyCMeansDistributedClusterer extends BaseFuzzyCMeansClusterer<SparseDistributedMatrix> {
+ /** Random numbers generator which is used in centers selection. */
+ private Random rnd;
+
+ /** The value that is used to initialize random numbers generator. */
+ private long seed;
+
+ /** The number of initialization steps each of which adds some number of candidates for being a center. */
+ private int initSteps;
+
+ /** The maximum number of iterations of K-Means algorithm which selects the required number of centers. */
+ private int kMeansMaxIterations;
+
+ /** The maximum number of FCM iterations. */
+ private int cMeansMaxIterations;
+
+ /**
+ * Constructor that retains all required parameters.
+ *
+ * @param measure Distance measure.
+ * @param exponentialWeight Specific constant which is used in calculating of membership matrix.
+ * @param stopCond Flag that tells when algorithm should stop.
+ * @param maxDelta The maximum distance between old and new centers or maximum difference between new and old
+ * membership matrix elements for which algorithm must stop.
+ * @param cMeansMaxIterations The maximum number of FCM iterations.
+ * @param seed Seed for random numbers generator.
+ * @param initSteps Number of steps of primary centers selection (the more steps, the more candidates).
+ * @param kMeansMaxIterations The maximum number of K-Means iteration in primary centers selection.
+ */
+ public FuzzyCMeansDistributedClusterer(DistanceMeasure measure, double exponentialWeight,
+ StopCondition stopCond, double maxDelta, int cMeansMaxIterations,
+ Long seed, int initSteps, int kMeansMaxIterations) {
+ super(measure, exponentialWeight, stopCond, maxDelta);
+
+ this.seed = seed != null ? seed : new Random().nextLong();
+ this.initSteps = initSteps;
+ this.cMeansMaxIterations = cMeansMaxIterations;
+ this.kMeansMaxIterations = kMeansMaxIterations;
+ rnd = new Random(this.seed);
+ }
+
+ /** {@inheritDoc} */
+ @Override public FuzzyCMeansModel cluster(SparseDistributedMatrix points, int k)
+ throws MathIllegalArgumentException, ConvergenceException {
+ GridArgumentCheck.notNull(points, "points");
+
+ if (k < 2)
+ throw new MathIllegalArgumentException("The number of clusters is less than 2");
+
+ Vector[] centers = initializeCenters(points, k);
+
+ MembershipsAndSums membershipsAndSums = null;
+
+ int iteration = 0;
+ boolean finished = false;
+ while (!finished && iteration < cMeansMaxIterations) {
+ MembershipsAndSums newMembershipsAndSums = calculateMembership(points, centers);
+ Vector[] newCenters = calculateNewCenters(points, newMembershipsAndSums, k);
+
+ if (stopCond == StopCondition.STABLE_CENTERS)
+ finished = isFinished(centers, newCenters);
+ else
+ finished = isFinished(membershipsAndSums, newMembershipsAndSums);
+
+ centers = newCenters;
+ membershipsAndSums = newMembershipsAndSums;
+
+ iteration++;
+ }
+
+ if (iteration == cMeansMaxIterations)
+ throw new ConvergenceException("Fuzzy C-Means algorithm has not converged after " +
+ Integer.toString(iteration) + " iterations");
+
+ return new FuzzyCMeansModel(centers, measure);
+ }
+
+ /**
+ * Choose k primary centers from source points.
+ *
+ * @param points Matrix with source points.
+ * @param k Number of centers.
+ * @return Array of primary centers.
+ */
+ private Vector[] initializeCenters(SparseDistributedMatrix points, int k) {
+ int pointsNum = points.rowSize();
+
+ Vector firstCenter = points.viewRow(rnd.nextInt(pointsNum));
+
+ List<Vector> centers = new ArrayList<>();
+ List<Vector> newCenters = new ArrayList<>();
+
+ centers.add(firstCenter);
+ newCenters.add(firstCenter);
+
+ ConcurrentHashMap<Integer, Double> costs = new ConcurrentHashMap<>();
+
+ int step = 0;
+ UUID uuid = points.getUUID();
+ String cacheName = ((SparseDistributedMatrixStorage) points.getStorage()).cacheName();
+
+ while(step < initSteps) {
+ ConcurrentHashMap<Integer, Double> newCosts = getNewCosts(cacheName, uuid, newCenters);
+
+ for (Integer key : newCosts.keySet())
+ costs.merge(key, newCosts.get(key), Math::min);
+
+ double costsSum = costs.values().stream().mapToDouble(Double::valueOf).sum();
+
+ newCenters = getNewCenters(cacheName, uuid, costs, costsSum, k);
+ centers.addAll(newCenters);
+
+ step++;
+ }
+
+ return chooseKCenters(cacheName, uuid, centers, k);
+ }
+
+ /**
+ * Calculate new distances from each point to the nearest center.
+ *
+ * @param cacheName Cache name of point matrix.
+ * @param uuid Uuid of point matrix.
+ * @param newCenters The list of centers that was added on previous step.
+ * @return Hash map with distances.
+ */
+ private ConcurrentHashMap<Integer, Double> getNewCosts(String cacheName, UUID uuid,
+ List<Vector> newCenters) {
+ return CacheUtils.distributedFold(cacheName,
+ (IgniteBiFunction<Cache.Entry<SparseMatrixKey, ConcurrentHashMap<Integer, Double>>,
+ ConcurrentHashMap<Integer, Double>,
+ ConcurrentHashMap<Integer, Double>>)(vectorWithIndex, map) -> {
+ Vector vector = VectorUtils.fromMap(vectorWithIndex.getValue(), false);
+
+ for (Vector center : newCenters)
+ map.merge(vectorWithIndex.getKey().index(), distance(vector, center), Functions.MIN);
+
+ return map;
+ },
+ key -> key.dataStructureId().equals(uuid),
+ (map1, map2) -> {
+ map1.putAll(map2);
+ return map1;
+ },
+ ConcurrentHashMap::new);
+ }
+
+ /**
+ * Choose some number of center candidates from source points according to their costs.
+ *
+ * @param cacheName Cache name of point matrix.
+ * @param uuid Uuid of point matrix.
+ * @param costs Hash map with costs (distances to nearest center).
+ * @param costsSum The sum of costs.
+ * @param k The estimated number of centers.
+ * @return The list of new candidates.
+ */
+ private List<Vector> getNewCenters(String cacheName, UUID uuid,
+ ConcurrentHashMap<Integer, Double> costs, double costsSum, int k) {
+ return CacheUtils.distributedFold(cacheName,
+ (IgniteBiFunction<Cache.Entry<SparseMatrixKey, Map<Integer, Double>>,
+ List<Vector>,
+ List<Vector>>)(vectorWithIndex, centers) -> {
+ Integer idx = vectorWithIndex.getKey().index();
+ Vector vector = VectorUtils.fromMap(vectorWithIndex.getValue(), false);
+
+ double probability = (costs.get(idx) * 2.0 * k) / costsSum;
+
+ if (rnd.nextDouble() < probability)
+ centers.add(vector);
+
+ return centers;
+ },
+ key -> key.dataStructureId().equals(uuid),
+ (list1, list2) -> {
+ list1.addAll(list2);
+ return list1;
+ },
+ ArrayList::new);
+ }
+
+ /**
+ * Weight candidates and use K-Means to choose required number of them.
+ *
+ * @param cacheName Cache name of the point matrix.
+ * @param uuid Uuid of the point matrix.
+ * @param centers The list of candidates.
+ * @param k The estimated number of centers.
+ * @return {@code k} centers.
+ */
+ private Vector[] chooseKCenters(String cacheName, UUID uuid, List<Vector> centers, int k) {
+ centers = centers.stream().distinct().collect(Collectors.toList());
+
+ ConcurrentHashMap<Integer, Integer> weightsMap = weightCenters(cacheName, uuid, centers);
+
+ List<Double> weights = new ArrayList<>(centers.size());
+
+ for (int i = 0; i < centers.size(); i++)
+ weights.add(i, Double.valueOf(weightsMap.getOrDefault(i, 0)));
+
+ DenseLocalOnHeapMatrix centersMatrix = MatrixUtil.fromList(centers, true);
+
+ KMeansLocalClusterer clusterer = new KMeansLocalClusterer(measure, kMeansMaxIterations, seed);
+ return clusterer.cluster(centersMatrix, k, weights).centers();
+ }
+
+ /**
+ * Weight each center with number of points for which this center is the nearest.
+ *
+ * @param cacheName Cache name of the point matrix.
+ * @param uuid Uuid of the point matrix.
+ * @param centers The list of centers.
+ * @return Hash map with weights.
+ */
+ public ConcurrentHashMap<Integer, Integer> weightCenters(String cacheName, UUID uuid, List<Vector> centers) {
+ if (centers.size() == 0)
+ return new ConcurrentHashMap<>();
+
+ return CacheUtils.distributedFold(cacheName,
+ (IgniteBiFunction<Cache.Entry<SparseMatrixKey, ConcurrentHashMap<Integer, Double>>,
+ ConcurrentHashMap<Integer, Integer>,
+ ConcurrentHashMap<Integer, Integer>>)(vectorWithIndex, counts) -> {
+ Vector vector = VectorUtils.fromMap(vectorWithIndex.getValue(), false);
+
+ int nearest = 0;
+ double minDistance = distance(centers.get(nearest), vector);
+
+ for (int i = 0; i < centers.size(); i++) {
+ double currDistance = distance(centers.get(i), vector);
+ if (currDistance < minDistance) {
+ minDistance = currDistance;
+ nearest = i;
+ }
+ }
+
+ counts.compute(nearest, (index, value) -> value == null ? 1 : value + 1);
+
+ return counts;
+ },
+ key -> key.dataStructureId().equals(uuid),
+ (map1, map2) -> {
+ map1.putAll(map2);
+ return map1;
+ },
+ ConcurrentHashMap::new);
+ }
+
+ /**
+ * Calculate matrix of membership coefficients for each point and each center.
+ *
+ * @param points Matrix with source points.
+ * @param centers Array of current centers.
+ * @return Membership matrix and sums of membership coefficients for each center.
+ */
+ private MembershipsAndSums calculateMembership(SparseDistributedMatrix points, Vector[] centers) {
+ String cacheName = ((SparseDistributedMatrixStorage) points.getStorage()).cacheName();
+ UUID uuid = points.getUUID();
+ double fuzzyMembershipCoefficient = 2 / (exponentialWeight - 1);
+
+ MembershipsAndSumsSupplier supplier = new MembershipsAndSumsSupplier(centers.length);
+
+ return CacheUtils.distributedFold(cacheName,
+ (IgniteBiFunction<Cache.Entry<SparseMatrixKey, ConcurrentHashMap<Integer, Double>>,
+ MembershipsAndSums,
+ MembershipsAndSums>)(vectorWithIndex, membershipsAndSums) -> {
+ Integer idx = vectorWithIndex.getKey().index();
+ Vector pnt = VectorUtils.fromMap(vectorWithIndex.getValue(), false);
+ Vector distances = new DenseLocalOnHeapVector(centers.length);
+ Vector pntMemberships = new DenseLocalOnHeapVector(centers.length);
+
+ for (int i = 0; i < centers.length; i++)
+ distances.setX(i, distance(centers[i], pnt));
+
+ for (int i = 0; i < centers.length; i++) {
+ double invertedFuzzyWeight = 0.0;
+
+ for (int j = 0; j < centers.length; j++) {
+ double val = Math.pow(distances.getX(i) / distances.getX(j), fuzzyMembershipCoefficient);
+ if (Double.isNaN(val))
+ val = 1.0;
+
+ invertedFuzzyWeight += val;
+ }
+
+ double membership = Math.pow(1.0 / invertedFuzzyWeight, exponentialWeight);
+ pntMemberships.setX(i, membership);
+ }
+
+ membershipsAndSums.memberships.put(idx, pntMemberships);
+ membershipsAndSums.membershipSums = membershipsAndSums.membershipSums.plus(pntMemberships);
+
+ return membershipsAndSums;
+ },
+ key -> key.dataStructureId().equals(uuid),
+ (mem1, mem2) -> {
+ mem1.merge(mem2);
+ return mem1;
+ },
+ supplier);
+ }
+
+ /**
+ * Calculate new centers according to membership matrix.
+ *
+ * @param points Matrix with source points.
+ * @param membershipsAndSums Membership matrix and sums of membership coefficient for each center.
+ * @param k The number of centers.
+ * @return Array of new centers.
+ */
+ private Vector[] calculateNewCenters(SparseDistributedMatrix points, MembershipsAndSums membershipsAndSums, int k) {
+ String cacheName = ((SparseDistributedMatrixStorage) points.getStorage()).cacheName();
+ UUID uuid = points.getUUID();
+
+ CentersArraySupplier supplier = new CentersArraySupplier(k, points.columnSize());
+
+ Vector[] centers = CacheUtils.distributedFold(cacheName,
+ (IgniteBiFunction<Cache.Entry<SparseMatrixKey, ConcurrentHashMap<Integer, Double>>,
+ Vector[],
+ Vector[]>)(vectorWithIndex, centerSums) -> {
+ Integer idx = vectorWithIndex.getKey().index();
+ Vector pnt = MatrixUtil.localCopyOf(VectorUtils.fromMap(vectorWithIndex.getValue(), false));
+ Vector pntMemberships = membershipsAndSums.memberships.get(idx);
+
+ for (int i = 0; i < k; i++) {
+ Vector weightedPnt = pnt.times(pntMemberships.getX(i));
+ centerSums[i] = centerSums[i].plus(weightedPnt);
+ }
+
+ return centerSums;
+ },
+ key -> key.dataStructureId().equals(uuid),
+ (sums1, sums2) -> {
+ for (int i = 0; i < k; i++)
+ sums1[i] = sums1[i].plus(sums2[i]);
+
+ return sums1;
+ },
+ supplier);
+
+ for (int i = 0; i < k; i++)
+ centers[i] = centers[i].divide(membershipsAndSums.membershipSums.getX(i));
+
+ return centers;
+ }
+
+ /**
+ * Check if centers have moved insignificantly.
+ *
+ * @param centers Old centers.
+ * @param newCenters New centers.
+ * @return The result of comparison.
+ */
+ private boolean isFinished(Vector[] centers, Vector[] newCenters) {
+ int numCenters = centers.length;
+
+ for (int i = 0; i < numCenters; i++)
+ if (distance(centers[i], newCenters[i]) > maxDelta)
+ return false;
+
+ return true;
+ }
+
+ /**
+ * Check memberships difference.
+ *
+ * @param membershipsAndSums Old memberships.
+ * @param newMembershipsAndSums New memberships.
+ * @return The result of comparison.
+ */
+ private boolean isFinished(MembershipsAndSums membershipsAndSums, MembershipsAndSums newMembershipsAndSums) {
+ if (membershipsAndSums == null)
+ return false;
+
+ double currMaxDelta = 0.0;
+ for (Integer key : membershipsAndSums.memberships.keySet()) {
+ double distance = measure.compute(membershipsAndSums.memberships.get(key),
+ newMembershipsAndSums.memberships.get(key));
+ if (distance > currMaxDelta)
+ currMaxDelta = distance;
+ }
+
+ return currMaxDelta <= maxDelta;
+ }
+
+ /** Service class used to optimize counting of membership sums. */
+ private class MembershipsAndSums {
+ /** Membership matrix. */
+ public ConcurrentHashMap<Integer, Vector> memberships = new ConcurrentHashMap<>();
+
+ /** Membership sums. */
+ public Vector membershipSums;
+
+ /**
+ * Default constructor.
+ *
+ * @param k The number of centers.
+ */
+ public MembershipsAndSums(int k) {
+ membershipSums = new DenseLocalOnHeapVector(k);
+ }
+
+ /**
+ * Merge results of calculation for different parts of points.
+ * @param another Another part of memberships and sums.
+ */
+ public void merge(MembershipsAndSums another) {
+ memberships.putAll(another.memberships);
+ membershipSums = membershipSums.plus(another.membershipSums);
+ }
+ }
+
+ /** Service class that is used to create new {@link MembershipsAndSums} instances. */
+ private class MembershipsAndSumsSupplier implements IgniteSupplier<MembershipsAndSums> {
+ /** The number of centers */
+ int k;
+
+ /**
+ * Constructor that retains the number of centers.
+ *
+ * @param k The number of centers.
+ */
+ public MembershipsAndSumsSupplier(int k) {
+ this.k = k;
+ }
+
+ /**
+ * Create new instance of {@link MembershipsAndSums}.
+ *
+ * @return {@link MembershipsAndSums} object.
+ */
+ @Override public MembershipsAndSums get() {
+ return new MembershipsAndSums(k);
+ }
+ }
+
+ /** Service class that is used to create new arrays of vectors. */
+ private class CentersArraySupplier implements IgniteSupplier<Vector[]> {
+ /** The number of centers. */
+ int k;
+
+ /** The number of coordinates. */
+ int dim;
+
+ /**
+ * Constructor that retains all required parameters.
+ *
+ * @param k The number of centers.
+ * @param dim The number of coordinates.
+ */
+ public CentersArraySupplier(int k, int dim) {
+ this.k = k;
+ this.dim = dim;
+ }
+
+ /**
+ * Create new array of vectors.
+ *
+ * @return Array of vectors.
+ */
+ @Override public Vector[] get() {
+ DenseLocalOnHeapVector[] centerSumsArr = new DenseLocalOnHeapVector[k];
+ for (int i = 0; i < k; i++)
+ centerSumsArr[i] = new DenseLocalOnHeapVector(dim);
+ return centerSumsArr;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClusterer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClusterer.java b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClusterer.java
new file mode 100644
index 0000000..1724da3
--- /dev/null
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClusterer.java
@@ -0,0 +1,254 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+import org.apache.ignite.internal.util.GridArgumentCheck;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.Matrix;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.exceptions.ConvergenceException;
+import org.apache.ignite.ml.math.exceptions.MathIllegalArgumentException;
+import org.apache.ignite.ml.math.impls.matrix.DenseLocalOnHeapMatrix;
+import org.apache.ignite.ml.math.impls.vector.DenseLocalOnHeapVector;
+
+/** Implements the local version of Fuzzy C-Means algorithm for weighted points. */
+public class FuzzyCMeansLocalClusterer extends BaseFuzzyCMeansClusterer<DenseLocalOnHeapMatrix> implements
+ WeightedClusterer<DenseLocalOnHeapMatrix, FuzzyCMeansModel> {
+ /** The maximum number of iterations. */
+ private int maxIterations;
+
+ /** The random numbers generator that is used to choose primary centers. */
+ private Random rnd;
+
+ /**
+ * Constructor that retains all required parameters.
+ *
+ * @param measure Distance measure.
+ * @param exponentialWeight Specific constant which is used in calculating of membership matrix.
+ * @param stopCond Flag that tells when algorithm should stop.
+ * @param maxDelta The maximum distance between old and new centers or maximum difference between new and old
+ * membership matrix elements for which algorithm must stop.
+ * @param maxIterations The maximum number of FCM iterations.
+ */
+ public FuzzyCMeansLocalClusterer(DistanceMeasure measure, double exponentialWeight, StopCondition stopCond,
+ double maxDelta, int maxIterations, Long seed) {
+ super(measure, exponentialWeight, stopCond, maxDelta);
+ this.maxIterations = maxIterations;
+ rnd = seed != null ? new Random(seed) : new Random();
+ }
+
+ /** {@inheritDoc} */
+ @Override public FuzzyCMeansModel cluster(DenseLocalOnHeapMatrix points, int k) {
+ List<Double> ones = new ArrayList<>(Collections.nCopies(points.rowSize(), 1.0));
+ return cluster(points, k, ones);
+ }
+
+ /** {@inheritDoc} */
+ @Override public FuzzyCMeansModel cluster(DenseLocalOnHeapMatrix points, int k, List<Double> weights)
+ throws MathIllegalArgumentException, ConvergenceException {
+ GridArgumentCheck.notNull(points, "points");
+ GridArgumentCheck.notNull(weights, "weights");
+
+ if (points.rowSize() != weights.size())
+ throw new MathIllegalArgumentException("The number of points and the number of weights are not equal");
+
+ if (k < 2)
+ throw new MathIllegalArgumentException("The number of clusters is less than 2");
+
+ Matrix centers = new DenseLocalOnHeapMatrix(k, points.columnSize());
+ Matrix distances = new DenseLocalOnHeapMatrix(k, points.rowSize());
+ Matrix membership = new DenseLocalOnHeapMatrix(k, points.rowSize());
+ Vector weightsVector = new DenseLocalOnHeapVector(weights.size());
+ for (int i = 0; i < weights.size(); i++)
+ weightsVector.setX(i, weights.get(i));
+
+ initializeCenters(centers, points, k, weightsVector);
+
+ int iteration = 0;
+ boolean finished = false;
+ while (iteration < maxIterations && !finished) {
+ calculateDistances(distances, points, centers);
+ Matrix newMembership = calculateMembership(distances, weightsVector);
+ Matrix newCenters = calculateNewCenters(points, newMembership);
+
+ if (this.stopCond == StopCondition.STABLE_CENTERS)
+ finished = areCentersStable(centers, newCenters);
+ else
+ finished = areMembershipStable(membership, newMembership);
+
+ centers = newCenters;
+ membership = newMembership;
+ iteration++;
+ }
+
+ if (iteration == maxIterations)
+ throw new ConvergenceException("Fuzzy C-Means algorithm has not converged after " +
+ Integer.toString(iteration) + " iterations");
+
+ Vector[] centersArr = new Vector[k];
+ for (int i = 0; i < k; i++)
+ centersArr[i] = centers.getRow(i);
+
+ return new FuzzyCMeansModel(centersArr, measure);
+ }
+
+ /**
+ * Choose {@code k} centers according to their weights.
+ *
+ * @param centers Output matrix containing primary centers.
+ * @param points Matrix of source points.
+ * @param k The number of centers.
+ * @param weights Vector of weights.
+ */
+ private void initializeCenters(Matrix centers, Matrix points, int k, Vector weights) {
+ //int dimensions = points.columnSize();
+ int numPoints = points.rowSize();
+
+ Vector firstCenter = points.viewRow(rnd.nextInt(numPoints));
+ centers.setRow(0, firstCenter.getStorage().data());
+
+ Vector costs = points.foldRows(vector -> distance(vector, firstCenter));
+ costs = costs.times(weights);
+
+ double sum = costs.sum();
+
+ for (int i = 1; i < k; i++) {
+ double probe = rnd.nextDouble() * sum;
+ double cntr = 0;
+ int id = 0;
+
+ for (int j = 0; j < numPoints; j++) {
+ cntr += costs.getX(j);
+ if (cntr >= probe) {
+ id = j;
+ break;
+ }
+ }
+
+ centers.setRow(i, points.viewRow(id).getStorage().data());
+ sum -= costs.get(id);
+ costs.set(id, 0.0);
+ }
+ }
+
+ /**
+ * Calculate matrix of distances form each point to each center.
+ *
+ * @param distances Output matrix.
+ * @param points Matrix that contains source points.
+ * @param centers Matrix that contains centers.
+ */
+ private void calculateDistances(Matrix distances, Matrix points, Matrix centers) {
+ int numPoints = points.rowSize();
+ int numCenters = centers.rowSize();
+
+ for (int i = 0; i < numCenters; i++)
+ for (int j = 0; j < numPoints; j++)
+ distances.set(i, j, distance(centers.viewRow(i), points.viewRow(j)));
+ }
+
+ /**
+ * Calculate membership matrix.
+ *
+ * @param distances Matrix of distances.
+ * @param weights Vector of weights.
+ * @
+ */
+ private Matrix calculateMembership(Matrix distances, Vector weights) {
+ Matrix newMembership = new DenseLocalOnHeapMatrix(distances.rowSize(), distances.columnSize());
+ int numPoints = distances.columnSize();
+ int numCenters = distances.rowSize();
+ double fuzzyMembershipCoefficient = 2 / (exponentialWeight - 1);
+
+ for (int i = 0; i < numCenters; i++) {
+ for (int j = 0; j < numPoints; j++) {
+ double invertedFuzzyWeight = 0.0;
+
+ for (int k = 0; k < numCenters; k++) {
+ double val = Math.pow(distances.get(i, j) / distances.get(k, j),
+ fuzzyMembershipCoefficient);
+ if (Double.isNaN(val))
+ val = 1.0;
+
+ invertedFuzzyWeight += val;
+ }
+
+ double weight = 1.0 / invertedFuzzyWeight * weights.getX(j);
+ newMembership.setX(i, j, Math.pow(weight, exponentialWeight));
+ }
+ }
+ return newMembership;
+ }
+
+ /**
+ * Calculate new centers using membership matrix.
+ *
+ * @param points Matrix of source points.
+ * @param membership Matrix that contains membership coefficients.
+ * @return Matrix that contains new centers.
+ */
+ private Matrix calculateNewCenters(Matrix points, Matrix membership) {
+ Vector membershipSums = membership.foldRows(Vector::sum);
+ Matrix newCenters = membership.times(points);
+
+ int numCenters = newCenters.rowSize();
+ for (int i = 0; i < numCenters; i++)
+ newCenters.viewRow(i).divide(membershipSums.getX(i));
+
+ return newCenters;
+ }
+
+ /**
+ * Check if centers have moved insignificantly.
+ *
+ * @param centers Old centers.
+ * @param newCenters New centers.
+ * @return The result of comparison.
+ */
+ private boolean areCentersStable(Matrix centers, Matrix newCenters) {
+ int numCenters = centers.rowSize();
+ for (int i = 0; i < numCenters; i++)
+ if (distance(centers.viewRow(i), newCenters.viewRow(i)) > maxDelta)
+ return false;
+
+ return true;
+ }
+
+ /**
+ * Check if membership matrix has changed insignificantly.
+ *
+ * @param membership Old membership matrix.
+ * @param newMembership New membership matrix.
+ * @return The result of comparison.
+ */
+ private boolean areMembershipStable(Matrix membership, Matrix newMembership) {
+ int numCenters = membership.rowSize();
+ int numPoints = membership.columnSize();
+
+ for (int i = 0; i < numCenters; i++)
+ for (int j = 0; j < numPoints; j++)
+ if (Math.abs(newMembership.getX(i, j) - membership.getX(i, j)) > maxDelta)
+ return false;
+
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansModel.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansModel.java b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansModel.java
new file mode 100644
index 0000000..41267b9
--- /dev/null
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/FuzzyCMeansModel.java
@@ -0,0 +1,88 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import java.util.Arrays;
+import org.apache.ignite.ml.Exportable;
+import org.apache.ignite.ml.Exporter;
+import org.apache.ignite.ml.FuzzyCMeansModelFormat;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.Vector;
+
+/** This class incapsulates result of clusterization. */
+public class FuzzyCMeansModel implements ClusterizationModel<Vector, Integer>, Exportable<FuzzyCMeansModelFormat> {
+ /** Centers of clusters. */
+ private Vector[] centers;
+
+ /** Distance measure. */
+ private DistanceMeasure measure;
+
+ /**
+ * Constructor that creates FCM model by centers and measure.
+ *
+ * @param centers Array of centers.
+ * @param measure Distance measure.
+ */
+ public FuzzyCMeansModel(Vector[] centers, DistanceMeasure measure) {
+ this.centers = Arrays.copyOf(centers, centers.length);
+ this.measure = measure;
+ }
+
+ /** Distance measure used while clusterization. */
+ public DistanceMeasure distanceMeasure() {
+ return measure;
+ }
+
+ /** @inheritDoc */
+ @Override public int clustersCount() {
+ return centers.length;
+ }
+
+ /** @inheritDoc */
+ @Override public Vector[] centers() {
+ return Arrays.copyOf(centers, centers.length);
+ }
+
+ /**
+ * Predict closest center index for a given vector.
+ *
+ * @param val Vector.
+ * @return Index of the closest center or -1 if it can't be found.
+ */
+ @Override public Integer predict(Vector val) {
+ int idx = -1;
+ double minDistance = Double.POSITIVE_INFINITY;
+
+ for (int i = 0; i < centers.length; i++) {
+ double currDistance = measure.compute(val, centers[i]);
+ if (currDistance < minDistance) {
+ minDistance = currDistance;
+ idx = i;
+ }
+ }
+
+ return idx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public <P> void saveModel(Exporter<FuzzyCMeansModelFormat, P> exporter, P path) {
+ FuzzyCMeansModelFormat mdlData = new FuzzyCMeansModelFormat(centers, measure);
+
+ exporter.save(mdlData, path);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/WeightedClusterer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/WeightedClusterer.java b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/WeightedClusterer.java
index 55fb359..1688087 100644
--- a/modules/ml/src/main/java/org/apache/ignite/ml/clustering/WeightedClusterer.java
+++ b/modules/ml/src/main/java/org/apache/ignite/ml/clustering/WeightedClusterer.java
@@ -33,6 +33,6 @@ public interface WeightedClusterer<P, M extends Model> extends Clusterer<P, M> {
* @param k count of centers.
* @param weights Weights.
*/
- public KMeansModel cluster(P points, int k, List<Double> weights) throws
+ public M cluster(P points, int k, List<Double> weights) throws
MathIllegalArgumentException, ConvergenceException;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/ClusteringTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/test/java/org/apache/ignite/ml/clustering/ClusteringTestSuite.java b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/ClusteringTestSuite.java
index b4cce5e..85f61fa 100644
--- a/modules/ml/src/test/java/org/apache/ignite/ml/clustering/ClusteringTestSuite.java
+++ b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/ClusteringTestSuite.java
@@ -27,7 +27,9 @@ import org.junit.runners.Suite;
@Suite.SuiteClasses({
KMeansDistributedClustererTestSingleNode.class,
KMeansDistributedClustererTestMultiNode.class,
- KMeansLocalClustererTest.class
+ KMeansLocalClustererTest.class,
+ FuzzyCMeansDistributedClustererTest.class,
+ FuzzyCMeansLocalClustererTest.class
})
public class ClusteringTestSuite {
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClustererTest.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClustererTest.java b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClustererTest.java
new file mode 100644
index 0000000..0cfa7b8
--- /dev/null
+++ b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansDistributedClustererTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Random;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.EuclideanDistance;
+import org.apache.ignite.ml.math.StorageConstants;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.impls.matrix.SparseDistributedMatrix;
+import org.apache.ignite.ml.math.impls.vector.DenseLocalOnHeapVector;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/** Tests that checks distributed Fuzzy C-Means clusterer. */
+public class FuzzyCMeansDistributedClustererTest extends GridCommonAbstractTest {
+ /** Number of nodes in grid. */
+ private static final int NODE_COUNT = 3;
+
+ /** Grid instance. */
+ private Ignite ignite;
+
+ /** Default constructor. */
+ public FuzzyCMeansDistributedClustererTest() {
+ super(false);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ ignite = grid(NODE_COUNT);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ for (int i = 1; i <= NODE_COUNT; i++)
+ startGrid(i);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /** Test that algorithm gives correct results on a small sample - 4 centers on the plane. */
+ public void testTwoDimensionsLittleData() {
+ IgniteUtils.setCurrentIgniteName(ignite.configuration().getIgniteInstanceName());
+
+ FuzzyCMeansDistributedClusterer clusterer = new FuzzyCMeansDistributedClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_MEMBERSHIPS,
+ 0.01, 500, null, 2, 50);
+
+ double[][] points = new double[][]{{-10, -10}, {-9, -11}, {-10, -9}, {-11, -9},
+ {10, 10}, {9, 11}, {10, 9}, {11, 9},
+ {-10, 10}, {-9, 11}, {-10, 9}, {-11, 9},
+ {10, -10}, {9, -11}, {10, -9}, {11, -9}};
+
+ SparseDistributedMatrix pntMatrix = new SparseDistributedMatrix(16, 2,
+ StorageConstants.ROW_STORAGE_MODE, StorageConstants.RANDOM_ACCESS_MODE);
+ for (int i = 0; i < 16; i++)
+ pntMatrix.setRow(i, points[i]);
+
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, 4);
+
+ Vector[] centers = mdl.centers();
+ Arrays.sort(centers, Comparator.comparing(vector -> Math.atan2(vector.get(1), vector.get(0))));
+
+ DistanceMeasure measure = mdl.distanceMeasure();
+
+ assertEquals(0, measure.compute(centers[0], new DenseLocalOnHeapVector(new double[]{-10, -10})), 1);
+ assertEquals(0, measure.compute(centers[1], new DenseLocalOnHeapVector(new double[]{10, -10})), 1);
+ assertEquals(0, measure.compute(centers[2], new DenseLocalOnHeapVector(new double[]{10, 10})), 1);
+ assertEquals(0, measure.compute(centers[3], new DenseLocalOnHeapVector(new double[]{-10, 10})), 1);
+ }
+
+ /** Perform N tests each of which contains M random points placed around K centers on the plane. */
+ public void testTwoDimensionsRandomlyPlacedPointsAndCenters() {
+ IgniteUtils.setCurrentIgniteName(ignite.configuration().getIgniteInstanceName());
+
+ final int numOfTests = 5;
+
+ final double exponentialWeight = 2.0;
+ final double maxCentersDelta = 0.01;
+ final int maxIterations = 500;
+ final Long seed = 1L;
+
+ DistanceMeasure measure = new EuclideanDistance();
+ FuzzyCMeansDistributedClusterer distributedClusterer = new FuzzyCMeansDistributedClusterer(measure,
+ exponentialWeight, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS,
+ maxCentersDelta, maxIterations, seed, 2, 50);
+
+ for (int i = 0; i < numOfTests; i++)
+ performRandomTest(distributedClusterer, i);
+ }
+
+ /**
+ * Test given clusterer on points placed randomly around vertexes of a regular polygon.
+ *
+ * @param distributedClusterer Tested clusterer.
+ * @param seed Seed for the random numbers generator.
+ */
+ public void performRandomTest(FuzzyCMeansDistributedClusterer distributedClusterer, long seed) {
+ final int minNumCenters = 2;
+ final int maxNumCenters = 5;
+ final double maxRadius = 1000;
+ final int maxPoints = 1000;
+ final int minPoints = 300;
+
+ Random random = new Random(seed);
+
+ int numCenters = random.nextInt(maxNumCenters - minNumCenters) + minNumCenters;
+
+ double[][] centers = new double[numCenters][2];
+
+ for (int i = 0; i < numCenters; i++) {
+ double radius = maxRadius;
+ double angle = Math.PI * 2.0 * i / numCenters;
+
+ centers[i][0] = Math.cos(angle) * radius;
+ centers[i][1] = Math.sin(angle) * radius;
+ }
+
+ int numPoints = minPoints + random.nextInt(maxPoints - minPoints);
+
+ double[][] points = new double[numPoints][2];
+
+ for (int i = 0; i < numPoints; i++) {
+ int center = random.nextInt(numCenters);
+ double randomDouble = random.nextDouble();
+ double radius = randomDouble * randomDouble * maxRadius / 10;
+ double angle = random.nextDouble() * Math.PI * 2.0;
+
+ points[i][0] = centers[center][0] + Math.cos(angle) * radius;
+ points[i][1] = centers[center][1] + Math.sin(angle) * radius;
+ }
+
+ SparseDistributedMatrix pntMatrix = new SparseDistributedMatrix(numPoints, 2,
+ StorageConstants.ROW_STORAGE_MODE, StorageConstants.RANDOM_ACCESS_MODE);
+
+ for (int i = 0; i < numPoints; i++)
+ pntMatrix.setRow(i, points[i]);
+
+ FuzzyCMeansModel mdl = distributedClusterer.cluster(pntMatrix, numCenters);
+ Vector[] computedCenters = mdl.centers();
+ DistanceMeasure measure = mdl.distanceMeasure();
+
+ int cntr = numCenters;
+
+ for (int i = 0; i < numCenters; i++) {
+ for (int j = 0; j < numCenters; j++) {
+ if (measure.compute(computedCenters[i], new DenseLocalOnHeapVector(centers[j])) < 100) {
+ cntr--;
+ break;
+ }
+ }
+ }
+
+ assertEquals(0, cntr);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/b13f088b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClustererTest.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClustererTest.java b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClustererTest.java
new file mode 100644
index 0000000..c58ffc7
--- /dev/null
+++ b/modules/ml/src/test/java/org/apache/ignite/ml/clustering/FuzzyCMeansLocalClustererTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.ignite.ml.clustering;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import org.apache.ignite.ml.math.DistanceMeasure;
+import org.apache.ignite.ml.math.EuclideanDistance;
+import org.apache.ignite.ml.math.Matrix;
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.math.exceptions.MathIllegalArgumentException;
+import org.apache.ignite.ml.math.impls.matrix.DenseLocalOnHeapMatrix;
+import org.apache.ignite.ml.math.impls.vector.DenseLocalOnHeapVector;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/** Tests that checks local Fuzzy C-Means clusterer. */
+public class FuzzyCMeansLocalClustererTest {
+ /** Test FCM on points that forms three clusters on the line. */
+ @Test
+ public void equalWeightsOneDimension() {
+ BaseFuzzyCMeansClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS,
+ 0.01, 10, null);
+
+ double[][] points = new double[][]{{-10}, {-9}, {-8}, {-7},
+ {7}, {8}, {9}, {10},
+ {-1}, {0}, {1}};
+
+ Matrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, 3);
+
+ Vector[] centers = mdl.centers();
+ Arrays.sort(centers, Comparator.comparing(vector -> vector.getX(0)));
+ assertEquals(-8.5, centers[0].getX(0), 2);
+ assertEquals(0, centers[1].getX(0), 2);
+ assertEquals(8.5, centers[2].getX(0), 2);
+ }
+
+ /** Test FCM on points that forms four clusters on the plane. */
+ @Test
+ public void equalWeightsTwoDimensions() {
+ BaseFuzzyCMeansClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS,
+ 0.01, 20, null);
+
+ double[][] points = new double[][]{{-10, -10}, {-9, -11}, {-10, -9}, {-11, -9},
+ {10, 10}, {9, 11}, {10, 9}, {11, 9},
+ {-10, 10}, {-9, 11}, {-10, 9}, {-11, 9},
+ {10, -10}, {9, -11}, {10, -9}, {11, -9}};
+
+ Matrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, 4);
+ Vector[] centers = mdl.centers();
+ Arrays.sort(centers, Comparator.comparing(vector -> Math.atan2(vector.get(1), vector.get(0))));
+
+ DistanceMeasure measure = mdl.distanceMeasure();
+
+ assertEquals(0, measure.compute(centers[0], new DenseLocalOnHeapVector(new double[]{-10, -10})), 1);
+ assertEquals(0, measure.compute(centers[1], new DenseLocalOnHeapVector(new double[]{10, -10})), 1);
+ assertEquals(0, measure.compute(centers[2], new DenseLocalOnHeapVector(new double[]{10, 10})), 1);
+ assertEquals(0, measure.compute(centers[3], new DenseLocalOnHeapVector(new double[]{-10, 10})), 1);
+ }
+
+ /** Test FCM on points which have the equal coordinates. */
+ @Test
+ public void checkCentersOfTheSamePointsTwoDimensions() {
+ BaseFuzzyCMeansClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_MEMBERSHIPS, 0.01, 10, null);
+
+ double[][] points = new double[][] {{3.3, 10}, {3.3, 10}, {3.3, 10}, {3.3, 10}, {3.3, 10}};
+
+ Matrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+
+ int k = 2;
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, k);
+ Vector exp = new DenseLocalOnHeapVector(new double[] {3.3, 10});
+ for (int i = 0; i < k; i++) {
+ Vector center = mdl.centers()[i];
+
+ for (int j = 0; j < 2; j++)
+ assertEquals(exp.getX(j), center.getX(j), 1);
+ }
+ }
+
+ /** Test FCM on points located on the circle. */
+ @Test
+ public void checkCentersLocationOnSphere() {
+ BaseFuzzyCMeansClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS, 0.01, 100, null);
+
+ int numOfPoints = 650;
+ double radius = 100.0;
+ double[][] points = new double [numOfPoints][2];
+
+ for (int i = 0; i < numOfPoints; i++) {
+ points[i][0] = Math.cos(Math.PI * 2 * i / numOfPoints) * radius;
+ points[i][1] = Math.sin(Math.PI * 2 * i / numOfPoints) * radius;
+ }
+
+ Matrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+
+ int k = 10;
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, k);
+
+ Vector sum = mdl.centers()[0];
+ for (int i = 1; i < k; i++)
+ sum = sum.plus(mdl.centers()[i]);
+
+ assertEquals(0, sum.kNorm(1), 1);
+ }
+
+ /** Test FCM on points that forms the line located on the plane. */
+ @Test
+ public void test2DLineClustering() {
+ BaseFuzzyCMeansClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS, 0.01, 50, null);
+
+ double[][] points = new double[][]{{1, 2}, {3, 6}, {5, 10}};
+
+ Matrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+
+ int k = 2;
+ FuzzyCMeansModel mdl = clusterer.cluster(pntMatrix, k);
+ Vector[] centers = mdl.centers();
+ Arrays.sort(centers, Comparator.comparing(vector -> vector.getX(0)));
+
+ Vector[] exp = {new DenseLocalOnHeapVector(new double[]{1.5, 3}),
+ new DenseLocalOnHeapVector(new double[]{4.5, 9})};
+
+ for (int i = 0; i < k; i++) {
+ Vector center = centers[i];
+
+ for (int j = 0; j < 2; j++)
+ assertEquals(exp[i].getX(j), center.getX(j), 0.5);
+ }
+ }
+
+ /** Test FCM on points that have different weights. */
+ @Test
+ public void differentWeightsOneDimension() {
+ FuzzyCMeansLocalClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS,
+ 0.01, 10, null);
+
+ double[][] points = new double[][]{{1}, {2}, {3}, {4}, {5}, {6}};
+
+ DenseLocalOnHeapMatrix pntMatrix = new DenseLocalOnHeapMatrix(points);
+ ArrayList<Double> weights = new ArrayList<>();
+ Collections.addAll(weights, 3.0, 2.0, 1.0, 1.0, 1.0, 1.0);
+
+ Vector[] centers1 = clusterer.cluster(pntMatrix, 2).centers();
+ Vector[] centers2 = clusterer.cluster(pntMatrix, 2, weights).centers();
+ Arrays.sort(centers1, Comparator.comparing(vector -> vector.getX(0)));
+ Arrays.sort(centers2, Comparator.comparing(vector -> vector.getX(0)));
+
+ assertTrue(centers1[0].get(0) - centers2[0].get(0) > 0.5);
+ }
+
+ /** Test FCM on illegal number of clusters. */
+ @Test(expected = MathIllegalArgumentException.class)
+ public void testIllegalNumberOfClusters() {
+ FuzzyCMeansLocalClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS, 0.01, 10, null);
+ double[][] points = new double[][]{{1}, {2}, {3}, {4}};
+
+ FuzzyCMeansModel cluster = clusterer.cluster(new DenseLocalOnHeapMatrix(points), 1);
+ }
+
+ /** Test FCM on different numbers of points and weights. */
+ @Test(expected = MathIllegalArgumentException.class)
+ public void testDifferentAmountsOfPointsAndWeights(){
+ FuzzyCMeansLocalClusterer clusterer = new FuzzyCMeansLocalClusterer(new EuclideanDistance(),
+ 2, BaseFuzzyCMeansClusterer.StopCondition.STABLE_CENTERS, 0.01, 10, null);
+ double[][] points = new double[][]{{1}, {2}, {3}, {4}};
+
+ ArrayList<Double> weights = new ArrayList<>();
+ Collections.addAll(weights, 1.0, 34.0, 2.5, 5.0, 0.5);
+
+ FuzzyCMeansModel cluster = clusterer.cluster(new DenseLocalOnHeapMatrix(points), 2, weights);
+ }
+}