编程导航会员会员
登录/注册
编程导航数据话题讨论

数据

64 参与
分享

快来分享你的内容吧～

写文章
写面经
向专家提问，帮你解惑
点击登录，快来和大家讨论吧～
表情
图片
话题
打卡
综合
交流
文章
问答

热门话题
更多>
1
训练营打卡
13920
2
学习打卡
6690
3
求职
2610
4
AI
2200
5
交流
1170
6
AI编程
600
7
Java
590
8
简历
520
9
编程
460
10
程序员
430

2025/7/1 数据 · 数据工程

## 数据工程 整体上可划分为3个阶段：数据采集、数据预处理和数据处理。数据处理又可分为两大方向：数据分析和数据挖掘，数据分析重在以统计和可视化手段获取对具体样本的特征概况，数据挖掘重在做预测/分类，可进一步分解为特征提取、模型训练、模型测试与评估三个步骤，完成这三种步骤存在三种模式：基于逻辑推理的规则方法、基于统计的机器学习和基于联结主义的深度学习。 1. 基于规则，就是人根据自己所感知到的人脑对事物的分析模式，亲自观察样本，寻找普遍模式进行预测或分类。 2. 基于机器学习，就是从海量数据中通过统计的相关数学方法寻找规律。 3. 基于深度学习，就是只提供基本数学函数结构，这个结构通过仿照人的神经元的激活模式来实现，再直接把数据放上去利用估计值和实际值的，使用微积分对整个函数中的参数进行调参，直到获得参数不再变化或变化幅度十分微小，就认为找到了规律。 它们的实际应用就以目前火热的计算机视觉与自然语言处理来举例。自然语言处理是专门处理诸如新闻报刊、名言名著、随笔短文等文字类的数据，应用有：搜索引擎、智能客服、机器翻译、deepseek；计算机视觉是专门处理诸如摄像、绘画、视频等图像类的数据，应用有：智能手机里通常会自带的图像去雾功能、美化功能、抠图、日常使用的看图识物、自动驾驶中的目标检测。你可以简单介绍一下数据分析和人工智能的概念和应用领域，如数据清洗、数据可视化、机器学习、自然语言处理等。同时，你也可以举个例子说明它们在实际应用中的应用，如数据分析可用于销售预测、用户行为分析等，人工智能可用于图像识别、语音识别、智能客服、聊天机器人等。

2025/7/1 数据 · 记录

人关注实际世界并予以记录。 ![image.png](https://pic.code-nav.cn/post_picture/1835857526080225281/evZGTqks7KA60wST.webp)

专线搭建、网络变更、数据迁移原理及逻辑

### 问题描述 支付机构（可理解为银行或第三方支付机构）与商户做线上支付业务，调用API支付接口后需要做专线搭建、网络变更（即配置防火墙将商户网络加入银行网络白名单）、数据迁移（迁移前还需要调用迁移接口），想问下这三步的整个流程逻辑以及背后原理，感谢 ### 背景信息 目前主要为商户提供支付接口、迁移接口以及申请防火墙开通申请 ### 具体疑问 专线搭建机制？会用到阿里云之类的 数据迁移机制？ 整个流程背后逻辑及原理 ### 已有理解 目前只是比较清楚业务顺序，商户方调用支付接口及迁移接口后，才可做测试，且专线搭建完成后，配置防火墙才可做数据迁移 ### 预期目标 希望可以得到详细解释并理解 ### 相关资料 提供与问题相关的代码片段、文档链接或者其他有助于理解的问题描述。请勿使用模糊不清的截图，可以使用代码块分享代码。 如果有多段代码，推荐使用 [代码小抄工具](https://codecopy.cn) 上传。

Canal 中间件速通

最近在做项目的时候碰到个没见过的东西，叫 Canal，赶紧学学。 # 概述 我们都知道对于一个系统，数据是非常重要的，数据存储的方式又很多元，MySQL 等关系型数据库、Elastic Search、HBase、Kafka、Redis等等。而 Canal 是阿里的一个开源组件，一个十多岁的老同志了，其作用在于：**同步数据库的增量数据到其他的存储应用**。 > Canal 地址 --> [https://github.com/alibaba/canal.git](https://github.com/alibaba/canal.git) ## Canal 定义 GitHub 上对 Canal 的介绍就一句话： > **canal [kə'næl]**，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。 解释成**大白话**就是：Canal 感知到 MySQL 数据变动，然后解析变动的数据，将变动的数据发送到 MQ、同步到其他数据库，并等待进一步业务逻辑处理。<br />再进一步解释**什么叫数据变动**呢？其实就是我们对 MySQL 做一些添加操作、删除操作、还有修改操作，这些操作都会引起我们的数据发生变化。<br />发生变化了之后，candle 它就感知到了，一感知到他把这些数据全部解析出来，放到要放的地方去，进而起到一个**数据同步**的作用。看官方给的示意图：<br /> <img src="https://pic.code-nav.cn/post_picture/1675780045722882049/iWWFyxIGMVQuqhSn.webp" alt="Canal.png" width="100%" /> ## 工作原理 ### MySQL 主从复制的原理 在理解 Canal 的工作原理之前，先温习一下八股：MySQL 的主从复制。<br />都说 MySQL 的日志比他的数据本身更重要。其中有一种日志是 binlog(binary log 即二进制日志文件) ，主要记录了 MySQL 数据库中数据的所有变化，即数据库执行的所有 DDL 和 DML 语句。因此，我们根据主库的 MySQL binlog 日志就能够将主库的数据同步到从库中，主从复制依赖于 MySQL 的 binlog。<br />该过程的示意图如下：<br /> <img src="https://pic.code-nav.cn/post_picture/1675780045722882049/4i3w3nNUq5i6kOYO.webp" alt="MySQL主从.png" width="100%" /> - 主库将数据库中数据的变化写入到 binlog - 从库连接主库 - 从库会创建一个 I/O 线程向主库请求更新的 binlog - 主库会创建一个 binlog dump 线程来发送 binlog ，从库中的 I/O 线程负责接收 - 从库的 I/O 线程将接收的 binlog 写入到 relay log 中。 - 从库的 SQL 线程读取 relay log 同步数据到本地（也就是再执行一遍 SQL ）。 ### Canal 的工作原理 从图中可以看出，Canal 的工作原理就是把自己伪装成MySQL slave（从库），模拟MySQL slave的交互协议向MySQL Mater发送 dump协议，MySQL mater收到canal发送过来的dump请求，开始推送binary log给canal，然后canal解析binary log，再发送到存储目的地，比如MySQL，Kafka，Elastic Search等等。 ## Canal 历史背景 [官网](https://github.com/alibaba/canal.git)上对 Canal 的历史背景有过一个简要的介绍，早期阿里巴巴在杭州美国美国都部署了一个机房，那这时候就存在着一个**跨机房数据同步**的业务需求。最初他们采用的方式是业务的触发（trigger），但这种方式操作起来不算方便，所以在 2010年的时候，他逐步使用了一个数据库解析这种方式来取代。那这种方式有个不好的点在于，出现了很多**数据库的增量订阅和消费**。<br />这也就是 Canal 诞生的背景，为了解决这一问题，大概在 2014年左右，在天猫双11的时候，就推出这个 Canal 组件，用来解决大型的促销活动中，MySQL 数据库的高并发，引起的读与写的一些问题，然后呢他又在阿里的整个集团内部进行推广和应用，并在 2017 年正式开源。<br />当前的 canal 支持源端 MySQL 版本包括 5.1.x ， 5.5.x ， 5.6.x ， 5.7.x ， 8.0.x # Canal 配置与启动 > 可参考官方的 QuickStart：[https://github.com/alibaba/canal/wiki/QuickStart](https://github.com/alibaba/canal/wiki/QuickStart) 这里假设已经安装好了 MySQL。 ## MySQL 准备 首先需要在 MySQL 终创建一个用户，并授权： ```sql -- 使用命令登录：mysql -u root -p -- 创建用户 用户名：canal 密码：canal create user 'canal'@'%' identified by 'canal'; -- 授权 *.*表示所有库 grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'canal'; ``` 检查 binlog 是否开启： ```shell mysql> show variables like 'log_bin'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin | ON | +---------------+-------+ 1 row in set, 1 warning (0.00 sec) ``` 若没有开启，则到配置文件`my.cnf`或者`my.ini`中 设置如下信息： ```properties [mysqld] # 启用 binlog log-bin=mysql-bin # 设置 server_id，用于标识 MySQL 实例，必须是一个唯一的整数 server-id=1 # 可选：设置 binlog 的格式，可以是 ROW、STATEMENT 或 MIXED binlog_format=ROW # 可选：指定 binlog 的保留天数，0 表示永远保留 expire_logs_days=7 # 可选：最大 binlog 文件大小 max_binlog_size=100M ``` 然后重启 MySQL 服务，再次验证： ```sql mysql> show variables like 'log_bin'; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin | ON | +---------------+-------+ 1 row in set, 1 warning (0.00 sec) mysql> show binary logs; +------------------+-----------+-----------+ | Log_name | File_size | Encrypted | +------------------+-----------+-----------+ | mysql-bin.000001 | 561 | No | +------------------+-----------+-----------+ 1 row in set (0.00 sec) mysql> show master status; +------------------+----------+--------------+------------------+-------------------+ | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | +------------------+----------+--------------+------------------+-------------------+ | mysql-bin.000001 | 561 | | | | +------------------+----------+--------------+------------------+-------------------+ 1 row in set (0.00 sec) ``` ## Canal 安装 > **不建议最新的版本，一般坑多** 从官网下载 Canal 的安装包：[https://github.com/alibaba/canal/releases](https://github.com/alibaba/canal/releases)<br /> <img src="https://pic.code-nav.cn/post_picture/1675780045722882049/blICxLwx4Hg3T4zK.webp" alt="Canal下载.png" width="100%" /> <br />对这几个包简单介绍： - **Canal.deployer** - 这是核心的 Canal 部署包，包含 Canal Server，负责监听 MySQL binlog 并将数据解析后推送到目标系统。大多数情况下，都需要这个包来启动 Canal。 - **Canal.adapter** - 这个组件是为了将 Canal 收集到的数据适配到不同的存储或消息队列中，比如 Kafka、Elasticsearch 等。如果需要将数据同步到特定的系统，则需要这个适配器。 - **Canal.admin** - 这是 Canal 的管理工具，提供了图形界面和 API，方便对 Canal 集群进行管理和监控。如果有多个 Canal 实例或者需要更方便的管理方式，则可以考虑安装这个组件。 - **Canal.example** - 是一个示例配置包，通常包含了一些典型的 Canal 配置示例和模板文件，帮助快速了解如何配置和使用 Canal。这个包的用途是为用户提供参考，包含一些常见场景的配置文件，比如 Canal 如何连接到 MySQL 主库、如何配置适配器将数据同步到 Kafka 或 Elasticsearch 等。你可以通过这些示例文件，快速搭建自己的 Canal 环境，或者根据示例调整自己的配置。 解压后，可以看到如下的文件夹：<br /> <img src="https://pic.code-nav.cn/post_picture/1675780045722882049/4K4wDvSv5Gg2wFqU.png" alt="Canal目录.png" width="100%" /> <br />进入到 conf 目录后，可看到如下的文件结构：<br /> <img src="https://pic.code-nav.cn/post_picture/1675780045722882049/RqLUfXPENjCbYkhZ.webp" alt="Canal配置目录.png" width="100%" /> <br />`canal.properties`中对 canal 的 Server 进行了一个整体的配置，其中有一段配置如下： ```properties ################################################# ######### destinations ############# ################################################# canal.destinations = example # conf root dir canal.conf.dir = ../conf # auto scan instance dir add/remove and start/stop instance canal.auto.scan = true canal.auto.scan.interval = 5 canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml ``` 这是用于指定 Canal 的`destinations`的，说简单点就是关于要监听哪个源的配置，上述的配置文件指定了监听 example 这个路径，从而会再具体去找 example 目录下的 `instance.properties` 配置文件，后者对应了具体的 MySQL 数据源。那么，如果是要监听 A 和 B 两个 MySQL 源，配置信息则应该写为<br />`canal.destinations = A, B`<br />，之后创建对应的 A、B 两个目录，在其中编写各自的`instance.properties`。 修改完上述配置后，进入到`conf.example`修改配置文件，只用改下面的几个配置，其他的保持默认即可： ```properties # binlog日志名称，与上一步配置一致 canal.instance.master.journal.name=mysql-bin.000001 # 与主库连接时起始的binlog偏移量 canal.instance.master.position=154 # MySQL数据解析表的黑名单，多个表用 , 隔开 canal.instance.filter.black.regex= # 数据库的用户名和密码，也可以设置常用的、root等 canal.instance.dbUsername=canal canal.instance.dbPassword=canal ``` ## Canal 启动 启动脚本位于`bin`目录下，如果是 Windows 环境，则需要简单修改启动脚本 start.bat，在下面这行中： ```shell set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%" ``` 去掉 `-Dlogback.configurationFile="%logback_configurationFile%"`。<br />然后双击启动脚本即可。 # Java 客户端操作 Canal 支持多种语言的客户端操作：<br /><img src="https://pic.code-nav.cn/post_picture/1675780045722882049/grNGkHcAjfj0yOQn.webp" alt="image.png" width="100%" /><br />此处以 Java 语言的操作为例。参考：[https://github.com/alibaba/canal/wiki/ClientExample](https://github.com/alibaba/canal/wiki/ClientExample) ## 依赖引入 从 Maven 仓库中拉取依赖： ```xml <!-- https://mvnrepository.com/artifact/com.alibaba.otter/canal.client --> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency> ``` ## 编写一个 Canal 客户端代码 接下来在一个 Spring Boot 的应用中编写一个简单的客户端代码，懒人选择直接 copy 官方示例代码😂 > 不一定非得是 Spring Boot 应用，普通的 Java 工程也可 ```java package com.liangshou.springbootdemo.canal; import java.net.InetSocketAddress; import java.util.List; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import com.alibaba.otter.canal.protocol.CanalEntry.EventType; import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.CanalEntry.RowData; import org.springframework.beans.factory.InitializingBean; import org.springframework.stereotype.Component; /** * @author X-L-S */ @Component public class CanalClient implements InitializingBean { private final static int BATCH_SIZE = 1000; @Override public void afterPropertiesSet() throws Exception { // 创建链接 CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress(AddressUtils.getHostAddress(), 11111), "example", "", ""); try { //打开连接 connector.connect(); //订阅数据库表,全部表 connector.subscribe(".*\\..*"); //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿 connector.rollback(); while (true) { // 获取指定数量的数据 Message message = connector.getWithoutAck(BATCH_SIZE); //获取批量ID long batchId = message.getId(); //获取批量的数量 int size = message.getEntries().size(); //如果没有数据 if (batchId == -1 || size == 0) { try { //线程休眠2秒 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } else { //如果有数据,处理数据 printEntry(message.getEntries()); } //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。 connector.ack(batchId); } } catch (Exception e) { e.printStackTrace(); } finally { connector.disconnect(); } } /** * 打印canal server解析binlog获得的实体类信息 */ private static void printEntry(List<Entry> entrys) { for (Entry entry : entrys) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { //开启/关闭事务的实体类型，跳过 continue; } //RowChange对象，包含了一行数据变化的所有特征 //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等 RowChange rowChage; try { rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } //获取操作类型：insert/update/delete类型 EventType eventType = rowChage.getEventType(); //打印Header信息 System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); //判断是否是DDL语句 if (rowChage.getIsDdl()) { System.out.println("================》;isDdl: true,sql:" + rowChage.getSql()); } //获取RowChange对象里的每一行数据，打印出来 for (RowData rowData : rowChage.getRowDatasList()) { //如果是删除语句 if (eventType == EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); //如果是新增语句 } else if (eventType == EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); //如果是更新的语句 } else { //变更前的数据 System.out.println("------->; before"); printColumn(rowData.getBeforeColumnsList()); //变更后的数据 System.out.println("------->; after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<Column> columns) { for (Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } } ``` 代码的功能比较粗暴，在控制台打印出 Canal 的信息。<br />之后编写一段建表的 SQL 语句： ```sql -- 创建库 create database if not exists canal_demo; -- 切换库 use canal_demo; -- 用户表 create table if not exists user ( id bigint auto_increment comment 'id' primary key, userAccount varchar(256) not null comment '账号', userPassword varchar(512) not null comment '密码', unionId varchar(256) null comment '微信开放平台id', mpOpenId varchar(256) null comment '公众号openId', userName varchar(256) null comment '用户昵称', userAvatar varchar(1024) null comment '用户头像', userProfile varchar(512) null comment '用户简介', userRole varchar(256) default 'user' not null comment '用户角色：user/admin/ban', createTime datetime default CURRENT_TIMESTAMP not null comment '创建时间', updateTime datetime default CURRENT_TIMESTAMP not null on update CURRENT_TIMESTAMP comment '更新时间', isDelete tinyint default 0 not null comment '是否删除', index idx_unionId (unionId) ) comment '用户' collate = utf8mb4_unicode_ci; ``` ## 简单结果演示 确保 MySQL 服务启动、Canal Server 启动、Java 应用启动，执行对数据库的操作，比如下面的插入一条数据： ```sql INSERT INTO user (userAccount, userPassword, unionId, mpOpenId, userName, userAvatar, userProfile, userRole) VALUES ('testAccount', 'testPassword', 'testUnionId', 'testMpOpenId', 'testUserName', 'http://example.com/avatar.jpg', 'This is a test user.', 'user'); ``` 即可在控制台中看到打印出来的 Canal 收集的信息： ```shell ================》; binlog[mysql-bin.000002:406] , name[canal_demo,user] , eventType : DELETE id : 1 update=false userAccount : testAccount update=false userPassword : testPassword update=false unionId : testUnionId update=false mpOpenId : testMpOpenId update=false userName : testUserName update=false userAvatar : http://example.com/avatar.jpg update=false userProfile : This is a test user. update=false userRole : user update=false createTime : 2024-08-31 17:05:58 update=false updateTime : 2024-08-31 17:05:58 update=false isDelete : 0 update=false ================》; binlog[mysql-bin.000002:878] , name[canal_demo,user] , eventType : INSERT id : 2 update=true userAccount : testAccount update=true userPassword : testPassword update=true unionId : testUnionId update=true mpOpenId : testMpOpenId update=true userName : testUserName update=true userAvatar : http://example.com/avatar.jpg update=true userProfile : This is a test user. update=true userRole : user update=true createTime : 2024-08-31 19:22:27 update=true updateTime : 2024-08-31 19:22:27 update=true isDelete : 0 update=true ``` # 未完待续……

《The Data Warehouse ETL Toolkit》（中文版）

阿里云盘下载地址 https://www.aliyundrive.com/s/KqgcJBD5Kgs 提取码: k96g 地址：https://www.aliyundrive.com/s/KqgcJBD5Kgs

实在智能RPA

地址：http://ai-indeed.com/

《阿里云云原生数据湖体系全解读》

地址：https://www.aliyundrive.com/s/WpTSHSy9VJ5

《干净的数据：数据清洗入门与实践》

地址：https://www.aliyundrive.com/s/fDTwToVQEwB

《鲜活的数据：数据可视化指南》

地址：https://www.aliyundrive.com/s/7rgopTTB9xh

编程零基础应当如何开始学习 Python ？

下面的内容由浅入深，建议按照先后顺序阅读学习。如果你觉得内容不错，希望别做松鼠党，收藏了就不了了之，坚持读完，有条件的最好是购买书籍阅读效果更好。本文分享的内容和文末推荐的书籍均出自vamei君。 地址：https://www.zhihu.com/question/20039623

下载 APP