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

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

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

行動框架

我們將通過實現(xiàn)一個簡單的管道來處理每個設備的數(shù)據(jù)，以測量給定區(qū)域的空氣質(zhì)量指數(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集群一起以無服務器的方式工作。 假設我們在Pulsar集群中傳輸空氣質(zhì)量指數(shù)，我們可以編寫一個函數(shù)來計算超出給定閾值的指數(shù)數(shù)量，并將結果寫回到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ù)管道卻展示了這些框架的運行方式，您可以以此為基礎來擴展您可能會進一步感興趣的每個框架中的知識。 您可以在此處查看完整的代碼示例。

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

總結

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

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