大數據挑戰

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

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

行動框架

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

阿帕奇火花

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

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; }

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

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

Apache Flink

與Spark相似，我們需要在Flink DataSet中表示數據，然后對其應用必要的轉換和操作：

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; }

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

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

阿帕奇風暴

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

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(); } }

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

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集群，這對于開發很方便，但是我們想將Storm拓撲提交到生產集群。 在這種情況下，我們將具有以下執行流程：

阿帕奇點燃

在Ignite中，我們需要先將數據放入分布式緩存中，然后再運行數據處理管道，該管道是在Ignite群集上以分布式方式執行的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群集中運行應用程序，它將具有以下執行流程：

榛樹噴射機

Hazelcast Jet在Hazelcast IMDG之上運行，并且與Ignite相似，如果我們要處理數據，我們需要首先將其放入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還提供集成而無需外部數據源，并且不需要將數據存儲在IMDG群集中。 您也可以在不首先將數據存儲到列表中的情況下進行聚合（查看Github中包含改進版本的完整示例）。 感謝Hazelcast工程團隊的Jaromir和Can的寶貴意見。

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

卡夫卡流

Kafka Streams是一個客戶端庫，使用Kafka主題作為數據處理管道的源和接收器。 為了在我們的方案中使用Kafka Streams庫，我們將把空氣質量指數數字放入數字 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應用程序實例將具有以下執行流程：

脈沖星函數

Apache Pulsar函數是輕量級的計算過程，可與Apache Pulsar集群一起以無服務器的方式工作。 假設我們在Pulsar集群中傳輸空氣質量指數，我們可以編寫一個函數來計算超出給定閾值的指數數量，并將結果寫回到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; } }

該函數以及Pulsar集群的執行流程如下：

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

摘要

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

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。