{"id":2024,"date":"2019-12-14T01:00:00","date_gmt":"2019-12-14T01:00:00","guid":{"rendered":"https:\/\/www.javaadvent.com\/?p=2024"},"modified":"2019-12-15T15:31:18","modified_gmt":"2019-12-15T15:31:18","slug":"popular-frameworks-for-big-data-processing-in-java","status":"publish","type":"post","link":"https:\/\/www.javaadvent.com\/2019\/12\/popular-frameworks-for-big-data-processing-in-java.html","title":{"rendered":"Popular frameworks for big data processing in Java"},"content":{"rendered":"\n
The big data challenge<\/h1>\n\n\n\n
The concept of big data is understood differently in the\nvariety of domains where companies face the need to deal with increasing\nvolumes of data. In most of these scenarios the system under consideration needs\nto be designed in such a way so that it is capable of processing that data without\nsacrificing throughput as data grows in size. This essentially leads to the necessity\nof building systems that are highly scalable so that more resources can be\nallocated based on the volume of data that needs to be processed at a given\npoint in time. <\/p>\n\n\n\n
Building such a system is a time-consuming and complex activity and for that reason a third-party frameworks and libraries can be used to provide the scalability requirements out of the box. There are already a number of good choices that can be used in Java applications and this article we will discuss briefly some of the most popular ones:<\/p>\n\n\n\n<\/figure>\n\n\n\n
The frameworks in action<\/h1>\n\n\n\n
We are going to demonstrate each of the frameworks by implementing\na simple pipeline for processing of data from devices that measure the air\nquality index for a given area. For simplicity we will assume that numeric data\nfrom the devices is either received in batches or in a streaming fashion. Throughout\nthe examples we are going to use the THRESHOLD constant to denote the value\nabove which we consider an area being polluted.<\/p>\n\n\n\n
Apache Spark<\/h2>\n\n\n\n
In Spark we need to first convert the data into a proper\nformat. We are going to use Datasets but we can also choose DataFrames or RDDs (Resilient\nDistributed Datasets) as an alternative for the data representation. We can\nthen apply a number of Spark transformations and actions in order to process\nthe data in a distributed fashion. <\/p>\n\n\n\n
\n
public long countPollutedRegions(String[] numbers) {\n\t\t\/\/ runs a Spark master that takes up 4 cores\n\t\tSparkSession session = SparkSession.builder().\n\t\t\t\tappName(\"AirQuality\").\n\t\t\t\tmaster(\"local[4]\").\n\t\t\t\tgetOrCreate();\n\t\t\/\/ converts the array of numbers to a Spark dataset\n\t\tDataset numbersSet = session.createDataset(Arrays.asList(numbers), \n\t\t\t\tEncoders.STRING());\n\t\t\n\t\t\/\/ runs the data pipeline on the local spark\n\t\tlong pollutedRegions = numbersSet.map(number -> Integer.valueOf(number), \n\t\t\t\tEncoders.INT())\n\t\t\t\t.filter(number -> number > THRESHOLD).count();\n\t\t\n\t\t\n\t\treturn pollutedRegions;\n\t}<\/pre>\n<\/div><\/div>\n\n\n\n
If we want to change the above application to read data from\nan external source, write to an external data source and run it on a Spark\ncluster rather than a local Spark instance we would have the following execution\nflow:<\/p>\n\n\n\n<\/figure>\n\n\n\n
The Spark driver might be either a separate instance or part\nof the Spark cluster.<\/p>\n\n\n\n
Apache Flink<\/h2>\n\n\n\n
Similarly to Spark we need to represent the data in a Flink\nDataSet and then apply the necessary transformations and actions over it:<\/p>\n\n\n\n
public long countPollutedRegions(String[] numbers) throws Exception {\n\t\t\/\/ creates a Flink execution environment with proper configuration\n\t\tStreamExecutionEnvironment env = StreamExecutionEnvironment.\n\t\t\t\tcreateLocalEnvironment();\n\n \t\/\/ converts the array of numbers to a Flink dataset and creates\n \t\/\/ the data pipiline\n\t\tDataStream stream = env.fromCollection(Arrays.asList(numbers)).\n\t\t\t\tmap(number -> Integer.valueOf(number))\n\t\t\t\t.filter(number -> number > THRESHOLD).returns(Integer.class);\n\t\tlong pollutedRegions = 0;\n\t\tIterator numbersIterator = DataStreamUtils.collect(stream);\n\t\twhile(numbersIterator.hasNext()) {\n\t\t\tpollutedRegions++;\n\t\t\tnumbersIterator.next();\n\t\t}\n\t\treturn pollutedRegions;\n\t}\n<\/pre>\n\n\n\n
If we want to change the above application to read data from\nan external source, write to an external data source and run it on a Flink\ncluster we would have the following execution flow:<\/p>\n\n\n\n<\/figure>\n\n\n\n
The Flink client where the application is submitted to the\nFlink cluster is either the Flink CLI utility or JobManager\u2019s UI.<\/p>\n\n\n\n
Apache Storm<\/h2>\n\n\n\n
In Storm the data pipeline is created as a topology of\nSpouts (the sources of data) and Bolts (the data processing units). Since Storm\ntypically processes unbounded streams of data we will emulate the processing of\nan array of air quality index numbers as bounded stream:<\/p>\n\n\n\n
\tpublic void countPollutedRegions(String[] numbers) throws Exception {\n\n\t\t\/\/ builds the topology as a combination of spouts and bolts\n\t\tTopologyBuilder builder = new TopologyBuilder();\n\t\tbuilder.setSpout(\"numbers-spout\", new StormAirQualitySpout(numbers));\n\t\tbuilder.setBolt(\"number-bolt\", new StormAirQualityBolt()).\n\t\t\tshuffleGrouping(\"numbers-spout\");\n\t\t\n\t\t\/\/ prepares Storm conf and along with the topology submits it for \n\t\t\/\/ execution to a local Storm cluster\n\t\tConfig conf = new Config();\n\t\tconf.setDebug(true);\n\t\tLocalCluster localCluster = null;\n\t\ttry {\n\t\t\tlocalCluster = new LocalCluster();\n\t\t\tlocalCluster.submitTopology(\"airquality-topology\", \n\t\t\t\t\tconf, builder.createTopology());\n\t\t\tThread.sleep(10000);\n\t\t\tlocalCluster.shutdown();\n\t\t} catch (InterruptedException ex) {\n\t\t\tlocalCluster.shutdown();\n\t\t}\n\t}\n<\/pre>\n\n\n\n
We have one spout that provides a data source for the array\nof air quality index numbers and one bolt that filters only the ones that indicate\npolluted areas: <\/p>\n\n\n\n
public class StormAirQualitySpout extends BaseRichSpout {\n\n\tprivate boolean emitted = false;\n\n\tprivate SpoutOutputCollector collector;\n\n\tprivate String[] numbers;\n\n\tpublic StormAirQualitySpout(String[] numbers) {\n\t\tthis.numbers = numbers;\n\t}\n\t\n\t@Override\n\tpublic void declareOutputFields(OutputFieldsDeclarer declarer) {\n\t\tdeclarer.declare(new Fields(\"number\"));\n\t}\n\n\t@Override\n\tpublic void open(Map paramas, \n\t\t\tTopologyContext context, \n\t\t\tSpoutOutputCollector collector) {\n\t\tthis.collector = collector;\n\t}\n\n\t@Override\n\tpublic void nextTuple() {\n\t\t\/\/ we make sure that the numbers array is processed just once by \n\t\t\/\/ the spout\n\t\tif(!emitted) {\n\t\t\tfor(String number : numbers) {\n\t\t\t\tcollector.emit(new Values(number));\n\t\t\t}\n\t\t\temitted = true;\n\t\t}\n\t}\n}\n<\/pre>\n\n\n\n
public class StormAirQualityBolt extends BaseRichBolt {\n\n\tprivate static final int THRESHOLD = 10;\n\n\tprivate int pollutedRegions = 0;\n\n\t@Override\n\tpublic void declareOutputFields(OutputFieldsDeclarer declarer) {\n\t\tdeclarer.declare(new Fields(\"number\"));\n\t}\n\n\t@Override\n\tpublic void prepare(Map params, \nTopologyContext context, \nOutputCollector collector) {\n\t}\n\n\t@Override\n\tpublic void execute(Tuple tuple) {\n\t\tString number = tuple.getStringByField(\"number\");\n\t\tInteger numberInt = Integer.valueOf(number);\n\t\tif (numberInt > THRESHOLD) {\n\t\t\tpollutedRegions++;\n\t\t}\n\t}\n}\n<\/pre>\n\n\n\n
We are using a LocalCluster instance for submitting to a\nlocal Storm cluster which is convenient for development purposes but we want to\nsubmit the Storm topology to a production cluster. In that case we would have\nthe following execution flow:<\/p>\n\n\n\n<\/figure>\n\n\n\n
Apache Ignite<\/h2>\n\n\n\n
In Ignite we need first to put the data in the distributed\ncache before running the data processing pipeline which is the former of an SQL\nquery executed in a distributed fashion over the Ignite cluster:<\/p>\n\n\n\n
\tpublic long countPollutedRegions(String[] numbers) {\n\n\t\tIgniteConfiguration igniteConfig = new IgniteConfiguration();\n\t\tCacheConfiguration cacheConfig = \n\t\t\t\tnew CacheConfiguration();\n\t\t\/\/ cache key is number index in the array and value is the number\n \tcacheConfig.setIndexedTypes(Integer.class, String.class);\n\n\t\tcacheConfig.setName(NUMBERS_CACHE);\n\t\tigniteConfig.setCacheConfiguration(cacheConfig);\n\t\t\n\t\ttry (Ignite ignite = Ignition.start(igniteConfig)) {\n\t\t\tIgniteCache cache = ignite.getOrCreateCache(NUMBERS_CACHE);\n\t\t\t\/\/ adds the numbers to the Ignite cache\n\t\t\ttry (IgniteDataStreamer streamer = \n\t\t\t\t\tignite.dataStreamer(cache.getName())) {\n\t\t\t\tint key = 0;\n\t\t\t\tfor (String number : numbers) {\n\t\t\t\t\tstreamer.addData(key++, number);\n\t\t\t\t}\n\t\t\t}\n\n\t\t\t\/\/ performs an SQL query over the cached numbers\n\t\t\tSqlFieldsQuery query = new SqlFieldsQuery(\"select * from String where _val > \" + THRESHOLD);\n\t\t\t\n\t\t\tFieldsQueryCursor<List> cursor = cache.query(query);\n\n\t\t\tint pollutedRegions = cursor.getAll().size();\n\t\t\treturn pollutedRegions;\n\t\t}\n}\n<\/pre>\n\n\n\n
If we want to run the application in an Ignite cluster it\nwill have the following execution flow:<\/p>\n\n\n\n<\/figure>\n\n\n\n
Hazelcast Jet<\/h2>\n\n\n\n
Hazelcast Jet works on top of Hazelcast IMDG and similarly\nto Ignite if we want to process data we need first to put it in the Hazelcast IMDG\ncluster:<\/p>\n\n\n\n
public long countPollutedRegions(String[] numbers) {\n\n\t\t\/\/ prepares the Jet data processing pipeline\n\t\tPipeline p = Pipeline.create();\n\t\tp.drawFrom(Sources.list(\"numbers\")).\n\t\t\tmap(number -> Integer.valueOf((String) number))\n\t\t\t.filter(number -> number > THRESHOLD).drainTo(Sinks.list(\"filteredNumbers\"));\n\n\t\tJetInstance jet = Jet.newJetInstance();\n\t\tIList numbersList = jet.getList(\"numbers\");\n\t\tnumbersList.addAll(Arrays.asList(numbers));\n\n\t\ttry {\n\t\t\t\/\/ submits the pipeline in the Jet cluster\n\t\t\tjet.newJob(p).join();\n\n\t\t\t\/\/ gets the filtered data from Hazelcast IMDG\n\t\t\tList filteredRecordsList = jet.getList(\"filteredNumbers\");\n\t\t\tint pollutedRegions = filteredRecordsList.size();\n\n\t\t\treturn pollutedRegions;\n\t\t} finally {\n\t\t\tJet.shutdownAll();\n\t\t}\n\t}\n<\/pre>\n\n\n\n
Note however that Jet also provides integration without of external data sources and data does not need to be stored in the IMDG cluster. You can also do the aggregation without first storing the data into a list (review the full example in Github that contains the improved version). Thanks to Jaromir and Can from Hazelcast engineering team for the valuable input. <\/p>\n\n\n\n
If we want to run the application in a Hazelcast Jet cluster it will have the following execution flow:<\/p>\n\n\n\n<\/figure>\n\n\n\n
Kafka Streams<\/h2>\n\n\n\n
Kafka Streams is a client library that uses Kafka topics as sources and sinks for the data processing pipeline. To make use of the Kafka Streams library for our scenario we would be putting the air quality index numbers in a numbers<\/strong> Kafka topic:<\/p>\n\n\n\n
public long countPollutedRegions() {\n\n\t\tList result = new LinkedList();\n\t\/\/ key\/value pairs contain string items\n \t\tfinal Serde stringSerde = Serdes.String();\n\n\t\t\/\/ prepares and runs the data processing pipeline\n\t\tfinal StreamsBuilder builder = new StreamsBuilder();\t\t\n\t\tbuilder.stream(\"numbers\", Consumed.with(stringSerde, stringSerde))\n\t\t\t\t.map((key, value) -> new KeyValue(key, Integer.valueOf(value))).\n\t\t\t\t\tfilter((key, value) -> value > THRESHOLD)\n\t\t\t\t.foreach((key, value) -> {\n\t\t\t\t\tresult.add(value.toString());\n\t\t\t\t});\n\t\n\t\tfinal Topology topology = builder.build();\n\t\tfinal KafkaStreams streams = new KafkaStreams(topology, \n\t\t\t\tcreateKafkaStreamsConfiguration());\n\t\tstreams.start();\n\n\t\ttry {\n\t\t\tThread.sleep(10000);\n\t\t} catch (InterruptedException e) {\n\t\t\te.printStackTrace();\n\t\t}\n\t\tint pollutedRegions = result.size();\n\t\tSystem.out.println(\"Number of severely polluted regions: \" + pollutedRegions);\n\t\tstreams.close();\n\t\treturn pollutedRegions;\n\t}\n\n\tprivate Properties createKafkaStreamsConfiguration() {\n\t\tProperties props = new Properties();\n\t\tprops.put(StreamsConfig.APPLICATION_ID_CONFIG, \"text-search-config\");\n\t\tprops.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, \"localhost:9092\");\n\t\tprops.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());\n\t\tprops.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());\n\n\t\treturn props;\n\t}\n<\/pre>\n\n\n\n
We will have the following execution flow for our Kafka Stream\napplication instances:<\/p>\n\n\n\n<\/figure>\n\n\n\n
Pulsar Functions <\/h2>\n\n\n\n
Apache Pulsar Functions are lightweight compute processes that work in a serverless fashion along with an Apache Pulsar cluster. Assuming we are streaming our air quality index in a Pulsar cluster we can write a function to count the number of indexes that exceed the given threshold and write the result back to Pulsar as follows:<\/p>\n\n\n\n
public class PulsarFunctionsAirQualityApplication \n\timplements Function {\n\n\tprivate static final int HIGH_THRESHOLD = 10;\n\n @Override\n public Void process(String input, Context context) throws Exception {\n \t\n \tint number = Integer.valueOf(input);\n \t\n \tif(number > HIGH_THRESHOLD) {\n context.incrCounter(\"pollutedRegions\", 1);\n \t}\n return null;\n }\n}<\/pre>\n\n\n\n
The execution flow of the function along with a Pulsar\ncluster is the following:<\/p>\n\n\n\n<\/figure>\n\n\n\n
The Pulsar function can run either in the Pulsar cluster or\nas a separate application.<\/p>\n\n\n\n
Summary<\/h1>\n\n\n\n
In this article we reviewed briefly some of the most popular frameworks that can be used to implement big data processing systems in Java. Each of the presented frameworks is fairly big and deserves a separate article on its own. Although quite simple our air quality index data pipeline demonstrates the way these frameworks operate and you can use that as a basis for expanding your knowledge in each one of them that might be of further interest. You can review the complete code samples here<\/a>.<\/p>\n","protected":false},"excerpt":{"rendered":"
![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/big_data_processing_frameworks.jpg?resize=547%2C301&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/spark.png?resize=600%2C330&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/flink.png?resize=600%2C294&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/storm.png?resize=600%2C254&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/ignite.png?resize=600%2C411&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/hazelcast.png?resize=600%2C423&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/kafka_streams.png?resize=600%2C376&ssl=1\")
<\/figure>\n\n\n\n![\"\"](\"https:\/\/i0.wp.com\/www.javaadvent.com\/content\/uploads\/2019\/12\/pulsar_functions.png?resize=557%2C397&ssl=1\")
<\/figure>\n\n\n\n