当前位置: 首页 > news >正文

大数据-玩转数据-Flink营销对账

一、说明

在电商网站中,订单的支付作为直接与营销收入挂钩的一环,在业务流程中非常重要。对于订单而言,为了正确控制业务流程,也为了增加用户的支付意愿,网站一般会设置一个支付失效时间,超过一段时间不支付的订单就会被取消。另外,对于订单的支付,我们还应保证用户支付的正确性,这可以通过第三方支付平台的交易数据来做一个实时对账。

二、思路

对于订单支付事件,用户支付完成其实并不算完,我们还得确认平台账户上是否到账了。而往往这会来自不同的日志信息,所以我们要同时读入两条流的数据来做合并处理。

三、数据准备

订单数据从OrderLog.csv中读取,交易数据从ReceiptLog.csv中读取
JavaBean类的准备

四、代码

package com.lyh.flink06;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;import java.util.HashMap;
import java.util.Map;public class Project_Order {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);SingleOutputStreamOperator<OrderEvent> orderEventString = env.readTextFile("input/OrderLog.csv").map(line -> {String[] data = line.split(",");return new OrderEvent(Long.valueOf(data[0]),data[1],data[2],Long.valueOf(data[3]));}).filter(log -> "pay".equals(log.getEventType()));SingleOutputStreamOperator<TxEvent> txEventString = env.readTextFile("input/ReceiptLog.csv").map(line -> {String[] data = line.split(",");return new TxEvent(data[0],data[1],Long.valueOf(data[2]));});orderEventString.connect(txEventString).keyBy(OrderEvent::getTxId,TxEvent::getTxId).process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {Map<String,OrderEvent> OrderEventmap = new HashMap<>();Map<String,TxEvent> TxEventmap = new HashMap<>();@Overridepublic void processElement1(OrderEvent value,Context ctx,Collector<String> out) throws Exception {TxEvent txEvent = TxEventmap.get(ctx.getCurrentKey());if (txEvent != null) {out.collect("订单" + value.getOrderId() + "对账成功");}else {OrderEventmap.put(ctx.getCurrentKey(),value);}}@Overridepublic void processElement2(TxEvent value,Context ctx,Collector<String> out) throws Exception {OrderEvent orderEvent = OrderEventmap.get(ctx.getCurrentKey());if (orderEvent != null) {out.collect("订单" + orderEvent.getOrderId() + "对账成功");}else {TxEventmap.put(ctx.getCurrentKey(),value);}}}).print();env.execute();}
}

五、结果

在这里插入图片描述

相关文章:

大数据-玩转数据-Flink营销对账

一、说明 在电商网站中&#xff0c;订单的支付作为直接与营销收入挂钩的一环&#xff0c;在业务流程中非常重要。对于订单而言&#xff0c;为了正确控制业务流程&#xff0c;也为了增加用户的支付意愿&#xff0c;网站一般会设置一个支付失效时间&#xff0c;超过一段时间不支…...

中英双语对话大语言模型:ChatGLM-6B

介绍 ChatGLM-6B 是一个开源的、支持中英双语的对话语言模型&#xff0c;基于 General Language Model (GLM) 架构&#xff0c;具有 62 亿参数。结合模型量化技术&#xff0c;用户可以在消费级的显卡上进行本地部署&#xff08;INT4 量化级别下最低只需 6GB 显存&#xff09;。…...

MES生产报工管理

一、MES生产报工管理的定义与功能&#xff1a; MES生产报工管理是指利用制造执行系统&#xff08;MES&#xff09;对生产过程进行实时监控、数据采集和分析&#xff0c;并及时记录和报告生产工单的实际完成情况。其主要功能包括&#xff1a; 1. 实时数据采集&#xff1a;通过…...

五、修改官方FreeRTOS例程(STM32F1)

1、官方源码下载 (1)进入FreeRTOS官网&#xff1a;FreeRTOS官网 (2)下载FreeRTOS。(选择带示例的下载) 2、删减目录 (1)下载后解压的FreeRTOS文件如下图所示。 (2)删除下图中红框勾选的文件。 FreeRTOS-Plus&#xff0c;FreeRTOS的生态文件&#xff0c;非必需的。tools&…...

pytorch基础实践-数据与预处理

文章目录 数据集Fashion-MNIST 数据集 数据预处理包的导入在Pytorch中进行 ETL利用torchvison包获取和处理数据集&#xff08;ET&#xff09; 访问数据集访问和查看 train_set 中的单个数据利用 DataLoader 成批访问数据 数据集 Fashion-MNIST 数据集 MNIST MNIST&#xff0c;…...

Java智慧工地系统源码(微服务+Java+Springcloud+Vue+MySQL)

智慧工地系统是依托物联网、互联网、AI、可视化建立的大数据管理平台&#xff0c;是一种全新的管理模式&#xff0c;能够实现劳务管理、安全施工、绿色施工的智能化和互联网化。围绕施工现场管理的人、机、料、法、环五大维度&#xff0c;以及施工过程管理的进度、质量、安全三…...

PV3D: A 3D GENERATIVE MODEL FOR PORTRAITVIDEO GENERATION 【2023 ICLR】

ICLR&#xff1a;International Conference on Learning Representations CCF-A 国际表征学习大会&#xff1a;深度学习的顶级会议 生成对抗网络(GANs)的最新进展已经证明了生成令人惊叹的逼真肖像图像的能力。虽然之前的一些工作已经将这种图像gan应用于无条件的2D人像视频生…...

Apache BeanUtils工具介绍

beanutils&#xff0c;顾名思义&#xff0c;是java bean的一个工具类&#xff0c;可以帮助我们方便的读取(get)和设置(set)bean属性值、动态定义和访问bean属性&#xff1b;细心的话&#xff0c;会发现其实JDK已经提供了一个java.beans包&#xff0c;同样可以实现以上功能&…...

java 原子操作 笔记

目录 java 变量原子操作 java byte[] 原子操作 java 变量原子操作 public class Counter {private int count 0;public synchronized void increment() {count;}public synchronized int getCount() {return count;} } java byte[] 原子操作 public class SharedArray {pr…...

什么是线程安全性问题?Java中有哪些常用的同步机制来解决线程安全性问题?

线程安全性问题是指在多线程环境下&#xff0c;多个线程同时访问和修改共享数据时可能引发的数据不一致、竞态条件和并发访问异常等问题。线程安全性问题的主要原因是多个线程之间的并发执行&#xff0c;导致数据的访问和修改顺序不确定&#xff0c;从而产生不一致的结果。 为…...

Gitlab 安装全流程

Version&#xff1a;gitlab-ce:16.2.4-ce.0 简介 Gitlab 是一个开源的 Git 代码仓库系统&#xff0c;可以实现自托管的 Github 项目&#xff0c;即用于构建私有的代码托管平台和项目管理系统。系统基于 Ruby on Rails 开发&#xff0c;速度快、安全稳定。它拥有与 Github 类似…...

pdf转word最简单方法~

pdf转word最简单方法&#xff01;pdf转word最简单方法我们都知道&#xff0c;PDF文件是一种只读文件格式&#xff0c;无法按照需求对PDF文件进行更改与编辑&#xff0c;从而影响到了PDF文件的使用。所以&#xff0c;我们需要将PDF文件转换为word文档&#xff0c;以此来保证文件…...

Android 9.0 WiFi 扫描结果上报和获取流程

本文是对wifi扫描结果上报和获取过程的java层代码流程梳理总结。 我们先分析扫描成功的上报和获取过程。 一、WiFi扫描成功的上报和获取过程 WiFi扫描成功的上报和获取大致是由三条不连贯流程组成的&#xff0c;分别是通知framework和WifiTracker获取扫描结果以及应用主动获取…...

Java 项目日志实例:Log4j2

点击下方关注我&#xff0c;然后右上角点击...“设为星标”&#xff0c;就能第一时间收到更新推送啦~~~ Apache Log4j 2 是对 Log4j 的升级&#xff0c;与其前身 Log4j 1.x 相比有了显着的改进&#xff0c;并提供了许多 Logback 可用的改进&#xff0c;同时支持 JCL 以及 SLF4J…...

Effective C++条款14——在资源管理类中小心coping行为(资源管理)

条款13导入这样的观念:“资源取得时机便是初始化时机”(Resource Acquisitionls Initialization; RAII)&#xff0c;并以此作为“资源管理类”的脊柱&#xff0c;也描述了auto_ ptr和tr1::shared ptr如何将这个观念表现在 heap-based资源上。然而并非所有资源都是heap-based&am…...

【网络教程】如何创建/添加钉钉机器人以及如何获取机器人的Token/Secret

文章目录 创建钉钉机器人添加钉钉机器人获取机器人的Token/Secret相关网站创建钉钉机器人 这里以PC端的操作为例,按照如下操作进行 访问 钉钉开放平台选择机器人选项卡,点击右上角的创建应用,这里会有一个弹窗,我这里选择的是继续使用旧版,如图按照要求填写相关信息创建自…...

wx原生微信小程序入门常用总结

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、定义值和修改值1、定义值2、修改值&#xff08;1&#xff09;代码&#xff08;2&#xff09;代码说明&#xff08;3&#xff09;注意点 二、点击事件三、微…...

制作一个专属于安防监控业的小程序商城

随着科技的发展和人们生活水平的提高&#xff0c;安防监控设备在我们的日常生活中起到了越来越重要的作用。因此&#xff0c;建立一个安防监控设备商城小程序就变得尤为重要。下面将介绍如何建立这样一个小程序。 第一步&#xff0c;登录乔拓云平台后台&#xff0c;进入商城管理…...

基于java羽毛球馆管理系统设计与实现

摘 要 时代的变化速度实在超出人类的所料&#xff0c;21世纪&#xff0c;计算机已经发展到各行各业&#xff0c;各个地区&#xff0c;它的载体媒介-计算机&#xff0c;大众称之为的电脑&#xff0c;是一种特高速的科学仪器&#xff0c;比人类的脑袋要灵光无数倍&#xff0c;什么…...

安装elasticsearch8.9.0及修改配置

安装es流程 打开文件,添加以下行 vim /etc/sysctl.conf vm.max_map_count=262144重启生效 sysctl -p创建用户 useradd es passwd es修改es目录所属用户 chown -R es:es /opt/elasticsearch-8.9.0如果内存不足,可以修改es的初始化内存和Max内存,修改文件/opt/elasticsearch-8…...

linux之kylin系统nginx的安装

一、nginx的作用 1.可做高性能的web服务器 直接处理静态资源&#xff08;HTML/CSS/图片等&#xff09;&#xff0c;响应速度远超传统服务器类似apache支持高并发连接 2.反向代理服务器 隐藏后端服务器IP地址&#xff0c;提高安全性 3.负载均衡服务器 支持多种策略分发流量…...

C++初阶-list的底层

目录 1.std::list实现的所有代码 2.list的简单介绍 2.1实现list的类 2.2_list_iterator的实现 2.2.1_list_iterator实现的原因和好处 2.2.2_list_iterator实现 2.3_list_node的实现 2.3.1. 避免递归的模板依赖 2.3.2. 内存布局一致性 2.3.3. 类型安全的替代方案 2.3.…...

Prompt Tuning、P-Tuning、Prefix Tuning的区别

一、Prompt Tuning、P-Tuning、Prefix Tuning的区别 1. Prompt Tuning(提示调优) 核心思想:固定预训练模型参数,仅学习额外的连续提示向量(通常是嵌入层的一部分)。实现方式:在输入文本前添加可训练的连续向量(软提示),模型只更新这些提示参数。优势:参数量少(仅提…...

基于uniapp+WebSocket实现聊天对话、消息监听、消息推送、聊天室等功能,多端兼容

基于 ​UniApp + WebSocket​实现多端兼容的实时通讯系统,涵盖WebSocket连接建立、消息收发机制、多端兼容性配置、消息实时监听等功能,适配​微信小程序、H5、Android、iOS等终端 目录 技术选型分析WebSocket协议优势UniApp跨平台特性WebSocket 基础实现连接管理消息收发连接…...

基于当前项目通过npm包形式暴露公共组件

1.package.sjon文件配置 其中xh-flowable就是暴露出去的npm包名 2.创建tpyes文件夹&#xff0c;并新增内容 3.创建package文件夹...

MMaDA: Multimodal Large Diffusion Language Models

CODE &#xff1a; https://github.com/Gen-Verse/MMaDA Abstract 我们介绍了一种新型的多模态扩散基础模型MMaDA&#xff0c;它被设计用于在文本推理、多模态理解和文本到图像生成等不同领域实现卓越的性能。该方法的特点是三个关键创新:(i) MMaDA采用统一的扩散架构&#xf…...

【SQL学习笔记1】增删改查+多表连接全解析(内附SQL免费在线练习工具)

可以使用Sqliteviz这个网站免费编写sql语句&#xff0c;它能够让用户直接在浏览器内练习SQL的语法&#xff0c;不需要安装任何软件。 链接如下&#xff1a; sqliteviz 注意&#xff1a; 在转写SQL语法时&#xff0c;关键字之间有一个特定的顺序&#xff0c;这个顺序会影响到…...

Module Federation 和 Native Federation 的比较

前言 Module Federation 是 Webpack 5 引入的微前端架构方案&#xff0c;允许不同独立构建的应用在运行时动态共享模块。 Native Federation 是 Angular 官方基于 Module Federation 理念实现的专为 Angular 优化的微前端方案。 概念解析 Module Federation (模块联邦) Modul…...

C++.OpenGL (10/64)基础光照(Basic Lighting)

基础光照(Basic Lighting) 冯氏光照模型(Phong Lighting Model) #mermaid-svg-GLdskXwWINxNGHso {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-GLdskXwWINxNGHso .error-icon{fill:#552222;}#mermaid-svg-GLd…...

【Java_EE】Spring MVC

目录 Spring Web MVC ​编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 ​编辑参数重命名 RequestParam ​编辑​编辑传递集合 RequestParam 传递JSON数据 ​编辑RequestBody ​…...