Java大数据处理的流行框架

大数据挑战

在公司需要处理不断增长的数据量的各个领域中，对大数据的概念有不同的理解。在大多数这些情况下，需要以某种方式设计所考虑的系统，以便能够处理该数据，而不会随着数据大小的增加而牺牲吞吐量。从本质上讲，这导致需要构建高度可伸缩的系统，以便可以根据在给定时间点需要处理的数据量来分配更多资源。

构建这样的系统是一项耗时且复杂的活动，因此，可以使用第三方框架和库来提供现成的可伸缩性要求。在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集群的执行流程如下：