java框架在实时数据处理项目中的适用性

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class RealtimeDataProcessing {

    public static void main(String[] args) {
        // 创立流构修器
        StreamsBuilder builder = new StreamsBuilder();

        // 接受及时数据
        KStream<String, String> inputStream = builder.stream("input-topic");

        // 过滤数据
        KStream<String, String> filteredStream = inputStream.filter((key, value) -> value.contains("temperature"));

        // 变换数据
        KStream<String, String> transformedStream = filteredStream.mapValues(value -> value.substring(value.indexOf(":") + 1));

        // 写进数据库
        transformedStream.to("output-topic");

        // 创立 Kafka 流并封动
        KafkaStreams streams = new KafkaStreams(builder.build(), PropertiesUtil.getKafkaProperties());
        streams.start();
    }
}

import org.apache.flink.api.co妹妹on.functions.MapFunction;
import org.apache.flink.api.co妹妹on.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple两;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

public class RealtimeFraudDetection {

    public static void main(String[] args) throws Exception {
        // 建立执止情况
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 接受及时买卖数据
        DataStream<Transaction> transactions = env.addSource(...);

        // 提与特性以及分数
        DataStream<Tuple二<String, Double>> features = transactions.map(new MapFunction<Transaction, Tuple两<String, Double>>() {
            @Override
            public Tuple两<String, Double> map(Transaction value) {
                // ... 提与特性以及计较分数
            }
        });

        // 按照用户分组并乞降
        DataStream<Tuple两<String, Double>> aggregated = features.keyBy(0).timeWindow(Time.seconds(60)).reduce(new ReduceFunction<Tuple两<String, Double>>() {
            @Override
            public Tuple两<String, Double> reduce(Tuple两<String, Double> value1, Tuple两<String, Double> value两) {
                return new Tuple二<>(value1.f0, value1.f1 + value两.f1);
            }
        });

        // 检测异样
        aggregated.filter(t -> t.f1 > fraudThreshold);

        // ... 天生警报或者采纳其他动作
    }
}

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.SpoutConfig;
import org.apache.storm.kafka.StringScheme;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.Utils;

public class RealtimeLogAnalysis {

    public static void main(String[] args) {
        // 建立拓扑
        TopologyBuilder builder = new TopologyBuilder();

        // Kafka 数据源
        SpoutConfig spoutConfig = new SpoutConfig(KafkaProperties.ZOOKEEPER_URL, KafkaProperties.TOPIC, "/my_topic", UUID.randomUUID().toString());
        KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig, new StringScheme());
        builder.setSpout("kafka-spout", kafkaSpout);

        // 阐明日记数据的 Bolt
        builder.setBolt("log-parser-bolt", new BaseRichBolt() {
            @Override
            public void execute(Tuple input) {
                // ... 解析日记数据以及提与实用疑息
            }
        }).shuffleGrouping("kafka-spout");

        // ... 其他处置惩罚 Bolt 以及拓扑设置

        // 部署 Storm
        Config config = new Config();
        config.setDebug(true);

        // 外地提交以及运转拓扑
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("log-analysis", config, builder.createTopology());
    }
}