Java大数据处理的流行框架
大數(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)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: junit编写测试代码_编写数据访问代码
- 下一篇: 安卓缩写英文(安卓缩写)