Distributed Computing
Ignite provides an API for distributing computations across cluster nodes in a balanced and fault-tolerant manner. You can submit individual tasks for execution as well as implement the MapReduce pattern with automatic task splitting. The API provides fine-grained control over the job distribution strategy.
Getting the Compute Interface
The main entry point for running distributed computations is the compute interface, which can be obtained from an instance of Ignite.
Ignite ignite = Ignition.start();
IgniteCompute compute = ignite.compute();var ignite = Ignition.Start();
var compute = ignite.GetCompute();Ignite ignite = Ignition::Start(cfg);
Compute compute = ignite.GetCompute();The compute interface provides methods for distributing different types of tasks over cluster nodes and running colocated computations.
Specifying the Set of Nodes for Computations
Each instance of the compute interface is associated with a set of nodes on which the tasks are executed.
When called without arguments, ignite.compute() returns the compute interface that is associated with all server nodes.
To obtain an instance for a specific subset of nodes, use Ignite.compute(ClusterGroup group).
In the following example, the compute interface is bound to the remote nodes only, i.e. all nodes except for the one that runs this code.
Ignite ignite = Ignition.start();
IgniteCompute compute = ignite.compute(ignite.cluster().forRemotes());var ignite = Ignition.Start();
var compute = ignite.GetCluster().ForRemotes().GetCompute();This API is not presently available for C++.Executing Tasks
Ignite provides three interfaces that can be implemented to represent a task and executed via the compute interface:
-
IgniteRunnable— an extension ofjava.lang.Runnablethat can be used to implement calculations that do not have input parameters and return no result. -
IgniteCallable— an extension ofjava.util.concurrent.Callablethat returns a specific value. -
IgniteClosure— a functional interface that accepts a parameter and returns a value.
You can execute a task once (on one of the nodes) or broadcast it to all nodes.
|
Important
|
In order to run tasks on the remote nodes, make sure the class definitions of the tasks are available on the nodes. You can do this in two ways:
|
Executing a Runnable Task
To execute a runnable task, use the run(…) method of the compute interface. The task is sent to one of the nodes associated with the compute instance.
IgniteCompute compute = ignite.compute();
// Iterate through all words and print
// each word on a different cluster node.
for (String word : "Print words on different cluster nodes".split(" ")) {
compute.run(() -> System.out.println(word));
}class PrintWordAction : IComputeAction
{
public void Invoke()
{
foreach (var s in "Print words on different cluster nodes".Split(" "))
{
Console.WriteLine(s);
}
}
}
public static void ComputeRunDemo()
{
var ignite = Ignition.Start(
new IgniteConfiguration
{
DiscoverySpi = new TcpDiscoverySpi
{
LocalPort = 48500,
LocalPortRange = 20,
IpFinder = new TcpDiscoveryStaticIpFinder
{
Endpoints = new[]
{
"127.0.0.1:48500..48520"
}
}
}
}
);
ignite.GetCompute().Run(new PrintWordAction());
}/*
* Function class.
*/
class PrintWord : public compute::ComputeFunc<void>
{
friend struct ignite::binary::BinaryType<PrintWord>;
public:
/*
* Default constructor.
*/
PrintWord()
{
// No-op.
}
/*
* Constructor.
*
* @param text Text.
*/
PrintWord(const std::string& word) :
word(word)
{
// No-op.
}
/**
* Callback.
*/
virtual void Call()
{
std::cout << word << std::endl;
}
/** Word to print. */
std::string word;
};
/**
* Binary type structure. Defines a set of functions required for type to be serialized and deserialized.
*/
namespace ignite
{
namespace binary
{
template<>
struct BinaryType<PrintWord>
{
static int32_t GetTypeId()
{
return GetBinaryStringHashCode("PrintWord");
}
static void GetTypeName(std::string& dst)
{
dst = "PrintWord";
}
static int32_t GetFieldId(const char* name)
{
return GetBinaryStringHashCode(name);
}
static int32_t GetHashCode(const PrintWord& obj)
{
return 0;
}
static bool IsNull(const PrintWord& obj)
{
return false;
}
static void GetNull(PrintWord& dst)
{
dst = PrintWord("");
}
static void Write(BinaryWriter& writer, const PrintWord& obj)
{
writer.RawWriter().WriteString(obj.word);
}
static void Read(BinaryReader& reader, PrintWord& dst)
{
dst.word = reader.RawReader().ReadString();
}
};
}
}
int main()
{
IgniteConfiguration cfg;
cfg.springCfgPath = "/path/to/configuration.xml";
Ignite ignite = Ignition::Start(cfg);
// Get binding instance.
IgniteBinding binding = ignite.GetBinding();
// Registering our class as a compute function.
binding.RegisterComputeFunc<PrintWord>();
// Get compute instance.
compute::Compute compute = ignite.GetCompute();
std::istringstream iss("Print words on different cluster nodes");
std::vector<std::string> words((std::istream_iterator<std::string>(iss)),
std::istream_iterator<std::string>());
// Iterate through all words and print
// each word on a different cluster node.
for (std::string word : words)
{
// Run compute task.
compute.Run(PrintWord(word));
}
}Executing a Callable Task
To execute a callable task, use the call(…) method of the compute interface.
Collection<IgniteCallable<Integer>> calls = new ArrayList<>();
// Iterate through all words in the sentence and create callable jobs.
for (String word : "How many characters".split(" "))
calls.add(word::length);
// Execute the collection of callables on the cluster.
Collection<Integer> res = ignite.compute().call(calls);
// Add all the word lengths received from cluster nodes.
int total = res.stream().mapToInt(Integer::intValue).sum();class CharCounter : IComputeFunc<int>
{
private readonly string arg;
public CharCounter(string arg)
{
this.arg = arg;
}
public int Invoke()
{
return arg.Length;
}
}
public static void ComputeFuncDemo()
{
var ignite = Ignition.Start(
new IgniteConfiguration
{
DiscoverySpi = new TcpDiscoverySpi
{
LocalPort = 48500,
LocalPortRange = 20,
IpFinder = new TcpDiscoveryStaticIpFinder
{
Endpoints = new[]
{
"127.0.0.1:48500..48520"
}
}
}
}
);
// Iterate through all words in the sentence and create callable jobs.
var calls = "How many characters".Split(" ").Select(s => new CharCounter(s)).ToList();
// Execute the collection of calls on the cluster.
var res = ignite.GetCompute().Call(calls);
// Add all the word lengths received from cluster nodes.
var total = res.Sum();
}/*
* Function class.
*/
class CountLength : public compute::ComputeFunc<int32_t>
{
friend struct ignite::binary::BinaryType<CountLength>;
public:
/*
* Default constructor.
*/
CountLength()
{
// No-op.
}
/*
* Constructor.
*
* @param text Text.
*/
CountLength(const std::string& word) :
word(word)
{
// No-op.
}
/**
* Callback.
* Counts number of characters in provided word.
*
* @return Word's length.
*/
virtual int32_t Call()
{
return word.length();
}
/** Word to print. */
std::string word;
};
/**
* Binary type structure. Defines a set of functions required for type to be serialized and deserialized.
*/
namespace ignite
{
namespace binary
{
template<>
struct BinaryType<CountLength>
{
static int32_t GetTypeId()
{
return GetBinaryStringHashCode("CountLength");
}
static void GetTypeName(std::string& dst)
{
dst = "CountLength";
}
static int32_t GetFieldId(const char* name)
{
return GetBinaryStringHashCode(name);
}
static int32_t GetHashCode(const CountLength& obj)
{
return 0;
}
static bool IsNull(const CountLength& obj)
{
return false;
}
static void GetNull(CountLength& dst)
{
dst = CountLength("");
}
static void Write(BinaryWriter& writer, const CountLength& obj)
{
writer.RawWriter().WriteString(obj.word);
}
static void Read(BinaryReader& reader, CountLength& dst)
{
dst.word = reader.RawReader().ReadString();
}
};
}
}
int main()
{
IgniteConfiguration cfg;
cfg.springCfgPath = "/path/to/configuration.xml";
Ignite ignite = Ignition::Start(cfg);
// Get binding instance.
IgniteBinding binding = ignite.GetBinding();
// Registering our class as a compute function.
binding.RegisterComputeFunc<CountLength>();
// Get compute instance.
compute::Compute compute = ignite.GetCompute();
std::istringstream iss("How many characters");
std::vector<std::string> words((std::istream_iterator<std::string>(iss)),
std::istream_iterator<std::string>());
int32_t total = 0;
// Iterate through all words in the sentence, create and call jobs.
for (std::string word : words)
{
// Add word length received from cluster node.
total += compute.Call<int32_t>(CountLength(word));
}
}Executing an IgniteClosure
To execute an IgniteClosure, use the apply(…) method of the compute interface. The method accepts a task and an input parameter for the task. The parameter is passed to the given IgniteClosure at the execution time.
IgniteCompute compute = ignite.compute();
// Execute closure on all cluster nodes.
Collection<Integer> res = compute.apply(String::length, Arrays.asList("How many characters".split(" ")));
// Add all the word lengths received from cluster nodes.
int total = res.stream().mapToInt(Integer::intValue).sum();class CharCountingFunc : IComputeFunc<string, int>
{
public int Invoke(string arg)
{
return arg.Length;
}
}
public static void Foo()
{
var ignite = Ignition.Start(
new IgniteConfiguration
{
DiscoverySpi = new TcpDiscoverySpi
{
LocalPort = 48500,
LocalPortRange = 20,
IpFinder = new TcpDiscoveryStaticIpFinder
{
Endpoints = new[]
{
"127.0.0.1:48500..48520"
}
}
}
}
);
var res = ignite.GetCompute().Apply(new CharCountingFunc(), "How many characters".Split());
int total = res.Sum();
}This API is not presently available for C++.Broadcasting a Task
The broadcast() method executes a task on all nodes associated with the compute instance.
// Limit broadcast to remote nodes only.
IgniteCompute compute = ignite.compute(ignite.cluster().forRemotes());
// Print out hello message on remote nodes in the cluster group.
compute.broadcast(() -> System.out.println("Hello Node: " + ignite.cluster().localNode().id()));class PrintNodeIdAction : IComputeAction
{
public void Invoke()
{
Console.WriteLine("Hello node: " +
Ignition.GetIgnite().GetCluster().GetLocalNode().Id);
}
}
public static void BroadcastDemo()
{
var ignite = Ignition.Start(
new IgniteConfiguration
{
DiscoverySpi = new TcpDiscoverySpi
{
LocalPort = 48500,
LocalPortRange = 20,
IpFinder = new TcpDiscoveryStaticIpFinder
{
Endpoints = new[]
{
"127.0.0.1:48500..48520"
}
}
}
}
);
// Limit broadcast to remote nodes only.
var compute = ignite.GetCluster().ForRemotes().GetCompute();
// Print out hello message on remote nodes in the cluster group.
compute.Broadcast(new PrintNodeIdAction());
}/*
* Function class.
*/
class Hello : public compute::ComputeFunc<void>
{
friend struct ignite::binary::BinaryType<Hello>;
public:
/*
* Default constructor.
*/
Hello()
{
// No-op.
}
/**
* Callback.
*/
virtual void Call()
{
std::cout << "Hello" << std::endl;
}
};
/**
* Binary type structure. Defines a set of functions required for type to be serialized and deserialized.
*/
namespace ignite
{
namespace binary
{
template<>
struct BinaryType<Hello>
{
static int32_t GetTypeId()
{
return GetBinaryStringHashCode("Hello");
}
static void GetTypeName(std::string& dst)
{
dst = "Hello";
}
static int32_t GetFieldId(const char* name)
{
return GetBinaryStringHashCode(name);
}
static int32_t GetHashCode(const Hello& obj)
{
return 0;
}
static bool IsNull(const Hello& obj)
{
return false;
}
static void GetNull(Hello& dst)
{
dst = Hello();
}
static void Write(BinaryWriter& writer, const Hello& obj)
{
// No-op.
}
static void Read(BinaryReader& reader, Hello& dst)
{
// No-op.
}
};
}
}
int main()
{
IgniteConfiguration cfg;
cfg.springCfgPath = "/path/to/configuration.xml";
Ignite ignite = Ignition::Start(cfg);
// Get binding instance.
IgniteBinding binding = ignite.GetBinding();
// Registering our class as a compute function.
binding.RegisterComputeFunc<Hello>();
// Broadcast to all nodes.
compute::Compute compute = ignite.GetCompute();
// Declaring function instance.
Hello hello;
// Print out hello message on nodes in the cluster group.
compute.Broadcast(hello);
}Asynchronous Execution
All methods described in the previous sections have asynchronous counterparts:
-
callAsync(…) -
runAsync(…) -
applyAsync(…) -
broadcastAsync(…)
The asynchronous methods return an IgniteFuture that represents the result of the operation. In the following example, a collection of callable tasks is executed asynchronously.
IgniteCompute compute = ignite.compute();
Collection<IgniteCallable<Integer>> calls = new ArrayList<>();
// Iterate through all words in the sentence and create callable jobs.
for (String word : "Count characters using a callable".split(" "))
calls.add(word::length);
IgniteFuture<Collection<Integer>> future = compute.callAsync(calls);
future.listen(fut -> {
// Total number of characters.
int total = fut.get().stream().mapToInt(Integer::intValue).sum();
System.out.println("Total number of characters: " + total);
});class CharCounter : IComputeFunc<int>
{
private readonly string arg;
public CharCounter(string arg)
{
this.arg = arg;
}
public int Invoke()
{
return arg.Length;
}
}
public static void AsyncDemo()
{
var ignite = Ignition.Start(
new IgniteConfiguration
{
DiscoverySpi = new TcpDiscoverySpi
{
LocalPort = 48500,
LocalPortRange = 20,
IpFinder = new TcpDiscoveryStaticIpFinder
{
Endpoints = new[]
{
"127.0.0.1:48500..48520"
}
}
}
}
);
var calls = "Count character using async compute"
.Split(" ").Select(s => new CharCounter(s)).ToList();
var future = ignite.GetCompute().CallAsync(calls);
future.ContinueWith(fut =>
{
var total = fut.Result.Sum();
Console.WriteLine("Total number of characters: " + total);
});
}/*
* Function class.
*/
class CountLength : public compute::ComputeFunc<int32_t>
{
friend struct ignite::binary::BinaryType<CountLength>;
public:
/*
* Default constructor.
*/
CountLength()
{
// No-op.
}
/*
* Constructor.
*
* @param text Text.
*/
CountLength(const std::string& word) :
word(word)
{
// No-op.
}
/**
* Callback.
* Counts number of characters in provided word.
*
* @return Word's length.
*/
virtual int32_t Call()
{
return word.length();
}
/** Word to print. */
std::string word;
};
/**
* Binary type structure. Defines a set of functions required for type to be serialized and deserialized.
*/
namespace ignite
{
namespace binary
{
template<>
struct BinaryType<CountLength>
{
static int32_t GetTypeId()
{
return GetBinaryStringHashCode("CountLength");
}
static void GetTypeName(std::string& dst)
{
dst = "CountLength";
}
static int32_t GetFieldId(const char* name)
{
return GetBinaryStringHashCode(name);
}
static int32_t GetHashCode(const CountLength& obj)
{
return 0;
}
static bool IsNull(const CountLength& obj)
{
return false;
}
static void GetNull(CountLength& dst)
{
dst = CountLength("");
}
static void Write(BinaryWriter& writer, const CountLength& obj)
{
writer.RawWriter().WriteString(obj.word);
}
static void Read(BinaryReader& reader, CountLength& dst)
{
dst.word = reader.RawReader().ReadString();
}
};
}
}
int main()
{
IgniteConfiguration cfg;
cfg.springCfgPath = "/path/to/configuration.xml";
Ignite ignite = Ignition::Start(cfg);
// Get binding instance.
IgniteBinding binding = ignite.GetBinding();
// Registering our class as a compute function.
binding.RegisterComputeFunc<CountLength>();
// Get compute instance.
compute::Compute asyncCompute = ignite.GetCompute();
std::istringstream iss("Count characters using callable");
std::vector<std::string> words((std::istream_iterator<std::string>(iss)),
std::istream_iterator<std::string>());
std::vector<Future<int32_t>> futures;
// Iterate through all words in the sentence, create and call jobs.
for (std::string word : words)
{
// Counting number of characters remotely.
futures.push_back(asyncCompute.CallAsync<int32_t>(CountLength(word)));
}
int32_t total = 0;
// Counting total number of characters.
for (Future<int32_t> future : futures)
{
// Waiting for results.
future.Wait();
total += future.GetValue();
}
// Printing result.
std::cout << "Total number of characters: " << total << std::endl;
}Task Execution Timeout
You can set a timeout for task execution. If the task does not finish within the given time frame, it be stopped and all jobs produced by this task are cancelled.
To execute a task with a timeout, use the withTimeout(…) method of the compute interface.
The method returns a compute interface that executes the first task given to it in a time-limited manner.
Consequent tasks do not have a timeout: you need to call withTimeout(…) for every task that should have a timeout.
IgniteCompute compute = ignite.compute();
compute.withTimeout(300_000).run(() -> {
// your computation
// ...
});Sharing State Between Jobs on Local Node
It is often useful to share a state between different compute jobs executed on one node. For this purpose, there is a shared concurrent local map available on each node.
IgniteCluster cluster = ignite.cluster();
ConcurrentMap<String, Integer> nodeLocalMap = cluster.nodeLocalMap();This API is not presently available for C#/.NET.This API is not presently available for C++.Node-local values are similar to thread local variables in that these values are not distributed and kept only on the local node. Node-local data can be used to share the state between compute jobs. It can also be used by deployed services.
In the following example, a job increments a node-local counter every time it executes on some node. As a result, the node-local counter on each node tells us how many times the job has executed on that node.
IgniteCallable<Long> job = new IgniteCallable<Long>() {
@IgniteInstanceResource
private Ignite ignite;
@Override
public Long call() {
// Get a reference to node local.
ConcurrentMap<String, AtomicLong> nodeLocalMap = ignite.cluster().nodeLocalMap();
AtomicLong cntr = nodeLocalMap.get("counter");
if (cntr == null) {
AtomicLong old = nodeLocalMap.putIfAbsent("counter", cntr = new AtomicLong());
if (old != null)
cntr = old;
}
return cntr.incrementAndGet();
}
};This API is not presently available for C#/.NET.This API is not presently available for C++.Accessing Data from Computational Tasks
If your computational task needs to access the data stored in caches, you can do it via the instance of Ignite:
public class MyCallableTask implements IgniteCallable<Integer> {
@IgniteInstanceResource
private Ignite ignite;
@Override
public Integer call() throws Exception {
IgniteCache<Long, Person> cache = ignite.cache("person");
// Get the data you need
Person person = cache.get(1L);
// do with the data what you need to do
return 1;
}
}class FuncWithDataAccess : IComputeFunc<int>
{
[InstanceResource] private IIgnite _ignite;
public int Invoke()
{
var cache = _ignite.GetCache<int, string>("someCache");
// get the data you need
string cached = cache.Get(1);
// do with data what you need to do, for example:
Console.WriteLine(cached);
return 1;
}
}/*
* Function class.
*/
class GetValue : public compute::ComputeFunc<void>
{
friend struct ignite::binary::BinaryType<GetValue>;
public:
/*
* Default constructor.
*/
GetValue()
{
// No-op.
}
/**
* Callback.
*/
virtual void Call()
{
Ignite& node = GetIgnite();
// Get the data you need
Cache<int64_t, Person> cache = node.GetCache<int64_t, Person>("person");
// do with the data what you need to do
Person person = cache.Get(1);
}
};Note that the example shown above may not be the most effective way.
The reason is that the person object that corresponds to key 1 may be located on a node that is different from the node where the task is executed.
In this case, the object is fetched via network. This can be avoided by colocating the task with the data.
|
Caution
|
If you want to use the key and value objects inside |
Apache, Apache Ignite, the Apache feather and the Apache Ignite logo are either registered trademarks or trademarks of The Apache Software Foundation.