大數(shù)據(jù)挑戰(zhàn)

在公司需要處理不斷增長的數(shù)據(jù)量的各個領(lǐng)域中，對大數(shù)據(jù)的概念有不同的理解。 在大多數(shù)這些情況下，需要以某種方式設(shè)計所考慮的系統(tǒng)，以便能夠處理該數(shù)據(jù)，而不會隨著數(shù)據(jù)大小的增加而犧牲吞吐量。 從本質(zhì)上講，這導致需要構(gòu)建高度可伸縮的系統(tǒng)，以便可以根據(jù)在給定時間點需要處理的數(shù)據(jù)量來分配更多資源。

構(gòu)建這樣的系統(tǒng)是一項耗時且復雜的活動，因此，可以使用第三方框架和庫來提供現(xiàn)成的可伸縮性要求。 在Java應用程序中已經(jīng)有很多不錯的選擇，本文將簡要討論一些最受歡迎的選擇：

行動框架

我們將通過實現(xiàn)一個簡單的管道來處理每個設(shè)備的數(shù)據(jù)，以測量給定區(qū)域的空氣質(zhì)量指數(shù)，從而演示每個框架。 為簡單起見，我們假定來自設(shè)備的數(shù)字數(shù)據(jù)是分批接收或以流方式接收的。 在整個示例中，我們將使用THRESHOLD常量表示該值，在該值之上，我們認為一個區(qū)域被污染。

阿帕奇火花

在Spark中，我們需要先將數(shù)據(jù)轉(zhuǎn)換為正確的格式。 我們將使用數(shù)據(jù)集，但我們也可以選擇數(shù)據(jù)幀或RDD（彈性分布式數(shù)據(jù)集）作為數(shù)據(jù)表示的替代方法。 然后，我們可以應用許多Spark轉(zhuǎn)換和操作，以便以分布式方式處理數(shù)據(jù)。

public long countPollutedRegions(String[] numbers) { // runs a Spark master that takes up 4 cores SparkSession session = SparkSession.builder(). appName( "AirQuality" ). master( "local[4]" ). getOrCreate(); // converts the array of numbers to a Spark dataset Dataset numbersSet = session.createDataset(Arrays.asList(numbers), Encoders.STRING()); ????????// runs the data pipeline on the local spark long pollutedRegions = numbersSet.map(number -> Integer.valueOf(number), Encoders. INT ()) .filter(number -> number > THRESHOLD).count(); ????????????????return pollutedRegions; }

如果要更改上述應用程序以從外部源讀取數(shù)據(jù)，寫入外部數(shù)據(jù)源并在Spark集群而不是本地Spark實例上運行，我們將具有以下執(zhí)行流程：

Spark驅(qū)動程序可以是單獨的實例，也可以是Spark群集的一部分。

Apache Flink

與Spark相似，我們需要在Flink DataSet中表示數(shù)據(jù)，然后對其應用必要的轉(zhuǎn)換和操作：

public long countPollutedRegions(String[] numbers) throws Exception { // creates a Flink execution environment with proper configuration StreamExecutionEnvironment env = StreamExecutionEnvironment. createLocalEnvironment(); // converts the array of numbers to a Flink dataset and creates // the data pipiline DataStream stream = env.fromCollection(Arrays.asList(numbers)). map(number -> Integer.valueOf(number)) .filter(number -> number > THRESHOLD).returns(Integer. class ); long pollutedRegions = 0; Iterator numbersIterator = DataStreamUtils.collect(stream); while (numbersIterator.hasNext()) { pollutedRegions++; numbersIterator.next(); } return pollutedRegions; }

如果要更改上述應用程序以從外部源讀取數(shù)據(jù)，寫入外部數(shù)據(jù)源并在Flink群集上運行，我們將具有以下執(zhí)行流程：

將應用程序提交到Flink群集的Flink客戶端是Flink CLI實用程序或JobManager的UI。

阿帕奇風暴

在Storm中，數(shù)據(jù)管道被創(chuàng)建為Spouts（數(shù)據(jù)源）和Bolts（數(shù)據(jù)處理單元）的拓撲。 由于Storm通常會處理無限制的數(shù)據(jù)流，因此我們會將空氣質(zhì)量指數(shù)編號數(shù)組的處理模擬為有限制的流：

public void countPollutedRegions(String[] numbers) throws Exception { // builds the topology as a combination of spouts and bolts TopologyBuilder builder = new TopologyBuilder(); builder.setSpout( "numbers-spout" StormAirQualitySpout(numbers)); "numbers-spout" , new StormAirQualitySpout(numbers)); builder.setBolt( "number-bolt" , new StormAirQualityBolt()). shuffleGrouping( "numbers-spout" shuffleGrouping( "numbers-spout" ); ????????// prepares Storm conf and along with the topology submits it for // execution to a local Storm cluster Config conf = new Config(); conf.setDebug( true ); LocalCluster localCluster = null; try { localCluster = new LocalCluster(); localCluster.submitTopology( "airquality-topology" , conf, builder.createTopology()); Thread.sleep(10000); localCluster.shutdown(); } catch (InterruptedException ex) { localCluster.shutdown(); } }

我們有一個噴嘴可以為空氣質(zhì)量指數(shù)編號的數(shù)組提供數(shù)據(jù)源，還有一個僅過濾指示污染區(qū)域的螺栓：

public class StormAirQualitySpout extends BaseRichSpout { private boolean emitted = false ; private SpoutOutputCollector collector; private String[] numbers; public StormAirQualitySpout(String[] numbers) { this .numbers = numbers; } ????@Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare( new Fields( "number" )); } @Override public void open(Map paramas, TopologyContext context, SpoutOutputCollector collector) { this .collector = collector; } @Override public void nextTuple() { // we make sure that the numbers array is processed just once by // the spout if (!emitted) { for (String number : numbers) { collector.emit( new Values(number)); } emitted = true ; } } } public class StormAirQualityBolt extends BaseRichBolt { private static final int THRESHOLD = 10; private int pollutedRegions = 0; @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare( new Fields( "number" )); } @Override public void prepare(Map params, TopologyContext context, OutputCollector collector) { } @Override public void execute(Tuple tuple) { String number = tuple.getStringByField( "number" ); Integer numberInt = Integer.valueOf(number); if (numberInt > THRESHOLD) { pollutedRegions++; } } }

我們正在使用LocalCluster實例提交到本地Storm集群，這對于開發(fā)很方便，但是我們想將Storm拓撲提交到生產(chǎn)集群。 在這種情況下，我們將具有以下執(zhí)行流程：

阿帕奇點燃

在Ignite中，我們需要先將數(shù)據(jù)放入分布式緩存中，然后再運行數(shù)據(jù)處理管道，該管道是在Ignite群集上以分布式方式執(zhí)行的SQL查詢的前者：

public long countPollutedRegions(String[] numbers) { IgniteConfiguration igniteConfig = new IgniteConfiguration(); CacheConfiguration cacheConfig = new CacheConfiguration(); // cache key is number index in the array and value is the number cacheConfig.setIndexedTypes(Integer. class , String. class ); cacheConfig.setName(NUMBERS_CACHE); igniteConfig.setCacheConfiguration(cacheConfig); ????????try (Ignite ignite = Ignition.start(igniteConfig)) { IgniteCache cache = ignite.getOrCreateCache(NUMBERS_CACHE); // adds the numbers to the Ignite cache try (IgniteDataStreamer streamer = ignite.dataStreamer(cache.getName())) { int key = 0; for (String number : numbers) { streamer.addData(key++, number); } } // performs an SQL query over the cached numbers SqlFieldsQuery query = new SqlFieldsQuery( "select * from String where _val > " + THRESHOLD); ????????????FieldsQueryCursor<List> cursor = cache.query(query); int pollutedRegions = cursor.getAll().size(); return pollutedRegions; } }

如果我們要在Ignite群集中運行應用程序，它將具有以下執(zhí)行流程：

榛樹噴射機

Hazelcast Jet在Hazelcast IMDG之上運行，并且與Ignite相似，如果我們要處理數(shù)據(jù)，我們需要首先將其放入Hazelcast IMDG群集中：

public long countPollutedRegions(String[] numbers) { // prepares the Jet data processing pipeline Pipeline p = Pipeline.create(); p.drawFrom(Sources.list( "numbers" )). map(number -> Integer.valueOf((String) number)) .filter(number -> number > THRESHOLD).drainTo(Sinks.list( "filteredNumbers" )); JetInstance jet = Jet.newJetInstance(); IList numbersList = jet.getList( "numbers" ); numbersList.addAll(Arrays.asList(numbers)); try { // submits the pipeline in the Jet cluster jet.newJob(p).join(); // gets the filtered data from Hazelcast IMDG List filteredRecordsList = jet.getList( "filteredNumbers" ); int pollutedRegions = filteredRecordsList.size(); return pollutedRegions; } finally { Jet.shutdownAll(); } }

但是請注意，Jet還提供集成而無需外部數(shù)據(jù)源，并且不需要將數(shù)據(jù)存儲在IMDG群集中。 您也可以在不首先將數(shù)據(jù)存儲到列表中的情況下進行聚合（查看Github中包含改進版本的完整示例）。 感謝Hazelcast工程團隊的Jaromir和Can的寶貴意見。

如果我們要在Hazelcast Jet集群中運行該應用程序，它將具有以下執(zhí)行流程：

卡夫卡流

Kafka Streams是一個客戶端庫，使用Kafka主題作為數(shù)據(jù)處理管道的源和接收器。 為了在我們的方案中使用Kafka Streams庫，我們將把空氣質(zhì)量指數(shù)數(shù)字放入數(shù)字 Kafka主題中：

public long countPollutedRegions() { List result = new LinkedList(); // key/value pairs contain string items final Serde stringSerde = Serdes.String(); // prepares and runs the data processing pipeline final StreamsBuilder builder = new StreamsBuilder(); builder.stream( "numbers" , Consumed.with(stringSerde, stringSerde)) .map((key, value) -> new KeyValue(key, Integer.valueOf(value))). filter((key, value) -> value > THRESHOLD) .foreach((key, value) -> { result.add(value.toString()); }); ????final Topology topology = builder.build(); final KafkaStreams streams = new KafkaStreams(topology, createKafkaStreamsConfiguration()); streams.start(); try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } int pollutedRegions = result.size(); System.out.println( "Number of severely polluted regions: " + pollutedRegions); streams.close(); return pollutedRegions; } private Properties createKafkaStreamsConfiguration() { Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "text-search-config" ); props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092" ); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); return props; }

我們的Kafka Stream應用程序?qū)嵗龑⒕哂幸韵聢?zhí)行流程：

脈沖星函數(shù)

Apache Pulsar函數(shù)是輕量級的計算過程，可與Apache Pulsar集群一起以無服務器的方式工作。 假設(shè)我們在Pulsar集群中傳輸空氣質(zhì)量指數(shù)，我們可以編寫一個函數(shù)來計算超出給定閾值的指數(shù)數(shù)量，并將結(jié)果寫回到Pulsar，如下所示：

public class PulsarFunctionsAirQualityApplication implements Function { private static final int HIGH_THRESHOLD = 10; @Override public Void process(String input, Context context) throws Exception { ????????int number = Integer.valueOf(input); ????????if (number > HIGH_THRESHOLD) { context.incrCounter( "pollutedRegions" , 1); } return null; } }

該函數(shù)以及Pulsar集群的執(zhí)行流程如下：

Pulsar函數(shù)可以在Pulsar群集中運行，也可以作為單獨的應用程序運行。

摘要

在本文中，我們簡要回顧了一些可用于在Java中實現(xiàn)大數(shù)據(jù)處理系統(tǒng)的最受歡迎的框架。 所提供的每個框架都相當大，值得單獨發(fā)表一篇文章。 盡管非常簡單，但我們的空氣質(zhì)量指數(shù)數(shù)據(jù)管道卻展示了這些框架的運行方式，您可以以此為基礎(chǔ)來擴展您可能會進一步感興趣的每個框架中的知識。 您可以在此處查看完整的代碼示例。

翻譯自: https://www.javacodegeeks.com/2019/12/popular-frameworks-for-big-data-processing-in-java.html

總結(jié)

以上是生活随笔為你收集整理的Java大数据处理的流行框架的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。