import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

public class FlinkWordCountExample {
    public static void main(String[] args) throws Exception {
        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 从本地端口 9999 读取文本流数据
        DataStreamSource<String> stream = env.socketTextStream("localhost", 9999);

        // 将每行文本按空格拆分，扁平化输出单词
        SingleOutputStreamOperator<String> words = stream.flatMap((String line, Collector<String> out) -> {
            Arrays.stream(line.split(" ")).forEach(out::collect);
        });

        // 将每个单词映射为 <单词, 1> 的键值对，方便后续统计
        SingleOutputStreamOperator<WordWithCount> wordCounts = words.map(word -> new WordWithCount(word, 1))
               .keyBy(WordWithCount::getWord) // 按单词进行分组
               .sum("count"); // 对每个分组内的计数进行求和

        // 打印统计结果
        wordCounts.print();
        // 执行流处理任务
        env.execute("Flink WordCount Example");
    }

    // 自定义类，用于存储单词和对应的计数
    public static class WordWithCount {
        private String word;
        private int count;

        public WordWithCount() {}

        public WordWithCount(String word, int count) {
            this.word = word;
            this.count = count;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public int getCount() {
            return count;
        }

        public void setCount(int count) {
            this.count = count;
        }

        @Override
        public String toString() {
            return "WordWithCount{" +
                    "word='" + word + '\'' +
                    ", count=" + count +
                    '}';
        }
    }
}

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import java.util.Map;
import java.util.Random;

public class StormExample {
    // 自定义 Spout，用于生成随机句子
    public static class RandomSentenceSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;
        // 预定义的句子数组
        private static final String[] sentences = {
            "the cow jumped over the moon",
            "an apple a day keeps the doctor away",
            "four score and seven years ago",
            "snow white and the seven dwarfs",
            "i am at two with nature"
        };
        private final Random random = new Random();

        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            // 初始化 Spout 输出收集器
            this.collector = collector;
        }

        @Override
        public void nextTuple() {
            // 每隔 100 毫秒生成一个随机句子
            Utils.sleep(100);
            String sentence = sentences[random.nextInt(sentences.length)];
            // 发射生成的句子
            collector.emit(new Values(sentence));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // 声明输出字段
            declarer.declare(new Fields("sentence"));
        }
    }

    // 自定义 Bolt，用于将句子拆分为单词
    public static class SplitSentenceBolt extends BaseBasicBolt {
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            // 获取输入的句子
            String sentence = tuple.getString(0);
            // 将句子按空格拆分为单词，并发射每个单词
            for (String word : sentence.split(" ")) {
                collector.emit(new Values(word));
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            // 声明输出字段
            declarer.declare(new Fields("word"));
        }
    }

    public static void main(String[] args) throws Exception {
        // 创建拓扑构建器
        TopologyBuilder builder = new TopologyBuilder();
        // 设置 Spout，并行度为 5
        builder.setSpout("spout", new RandomSentenceSpout(), 5);
        // 设置 Bolt，并行度为 8，通过随机分组从 Spout 接收数据
        builder.setBolt("split", new SplitSentenceBolt(), 8).shuffleGrouping("spout");
        // 创建配置对象
        Config conf = new Config();
        // 开启调试模式
        conf.setDebug(true);
        // 创建本地集群
        LocalCluster cluster = new LocalCluster();
        // 提交拓扑到集群
        cluster.submitTopology("Getting-Started-Topology", conf, builder.createTopology());
        // 运行 10 秒
        Utils.sleep(10000);
        // 杀死拓扑
        cluster.killTopology("Getting-Started-Topology");
        // 关闭集群
        cluster.shutdown();
    }
}

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer
import scala.collection.mutable.HashMap
import redis.clients.jedis.Jedis

object EcommerceAnalysis {
  def main(args: Array[String]) {
    // 创建 Spark 配置对象
    val sparkConf = new SparkConf().setAppName("EcommerceAnalysis").setMaster("local[2]")
    // 创建流上下文，批处理间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    // 配置 Kafka 参数
    val kafkaParams = HashMap[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test-group",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    // 要订阅的 Kafka 主题
    val topics = Array("ecommerce-data")
    // 创建从 Kafka 读取数据的直接流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )
    // 提取流中的值（即消息内容）
    val lines = stream.map(_.value)
    // 数据处理逻辑
    val processedData = lines.map(line => {
      // 假设数据格式为 "user_id,product_id,action"
      val fields = line.split(",")
      val userId = fields(0)
      val productId = fields(1)
      val action = fields(2)
      // 进行简单的数据处理，如统计用户的购买次数
      if (action == "purchase") {
        (userId, 1)
      } else {
        (userId, 0)
      }
    }).reduceByKey(_ + _)
    // 将处理结果存储到 Redis 中
    processedData.foreachRDD(rdd => {
      rdd.foreachPartition(partition => {
        val jedis = new Jedis("localhost", 6379)
        partition.foreach { case (userId, count) =>
          jedis.set(userId, count.toString)
        }
        jedis.close()
      })
    })
    // 启动流处理
    ssc.start()
    // 等待流处理结束
    ssc.awaitTermination()
  }
}

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class FinancialRiskMonitoring {
  public static void main(String[] args) throws Exception {
    // 获取流执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 从消息队列中读取交易数据，这里简单模拟
    DataStream<String> transactions = env.socketTextStream("localhost", 9999);
    // 解析交易数据，假设数据格式为 "transaction_id,user_id,amount"
    DataStream<Transaction> transactionStream = transactions.map(line => {
      String[] fields = line.split(",");
      String transactionId = fields[0];
      String userId = fields[1];
      double amount = Double.parseDouble(fields[2]);
      return new Transaction(transactionId, userId, amount);
    });
    // 按用户 ID 分组，统计每个用户在 5 分钟内的交易总金额
    DataStream<TransactionSummary> summaryStream = transactionStream
      .keyBy(Transaction::getUserId)
      .timeWindow(Time.minutes(5))
      .process(new TransactionSummaryProcessFunction());
    // 检测异常交易
    DataStream<Transaction> abnormalTransactions = summaryStream
      .flatMap((TransactionSummary summary, Collector<Transaction> out) -> {
        if (summary.getTotalAmount() > 10000) {
          // 假设交易总金额超过 10000 为异常交易
          for (Transaction transaction : summary.getTransactions()) {
            out.collect(transaction);
          }
        }
      });
    // 输出异常交易信息
    abnormalTransactions.print();
    // 执行流处理任务
    env.execute("Financial Risk Monitoring");
  }

  // 交易类
  public static class Transaction {
    private String transactionId;
    private String userId;
    private double amount;

    public Transaction(String transactionId, String userId, double amount) {
      this.transactionId = transactionId;
      this.userId = userId;
      this.amount = amount;
    }

    public String getTransactionId() {
      return transactionId;
    }

    public String getUserId() {
      return userId;
    }

    public double getAmount() {
      return amount;
    }

    @Override
    public String toString() {
      return "Transaction{" +
        "transactionId='" + transactionId + '\'' +
        ", userId='" + userId + '\'' +
        ", amount=" + amount +
        '}';
    }
  }

  // 交易汇总类
  public static class TransactionSummary {
    private String userId;
    private double totalAmount;
    private List<Transaction> transactions;

    public TransactionSummary(String userId, double totalAmount, List<Transaction> transactions) {
      this.userId = userId;
      this.totalAmount = totalAmount;
      this.transactions = transactions;
    }

    public String getUserId() {
      return userId;
    }

    public double getTotalAmount() {
      return totalAmount;
    }

    public List<Transaction> getTransactions() {
      return transactions;
    }
  }

  // 处理窗口函数，统计交易总金额
  public static class TransactionSummaryProcessFunction extends ProcessWindowFunction<Transaction, TransactionSummary, String, TimeWindow> {
    @Override
    public void process(String userId, Context context, Iterable<Transaction> elements, Collector<TransactionSummary> out) throws Exception {
      double totalAmount = 0;
      List<Transaction> transactions = new ArrayList<>();
      for (Transaction transaction : elements) {
        totalAmount += transaction.getAmount();
        transactions.add(transaction);
      }
      out.collect(new TransactionSummary(userId, totalAmount, transactions));
    }
  }
}