我要评论先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里p7
深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！
因此收集整理了一份《2024年最新大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。
既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！
由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新
如果你需要这些资料，可以添加v获取：vip204888 （备注大数据）
正文
开启mysql数据库bin-log日志
1.如果是服务器
在my.cnf中添加binlog配置,并重启mysql数据库
server-id = 123
log_bin = mysql-bin
binlog_format = row
binlog_row_image = full
expire_logs_days = 10
gtid_mode = on
enforce_gtid_consistency = on
已经为 mysql 设置了一些配置参数。下面是对这些参数的解释：
- server-id = 123：指定服务器的唯一标识符，通常用于区分不同的数据库服务器。
- log_bin = mysql-bin：启用二进制日志记录，以便在数据库出现故障时可以恢复数据。
- binlog_format = row：指定二进制日志的记录格式。row 格式会记录每个更改行的详细信息，这对于需要事务完整性的应用程序非常有用。
- binlog_row_image = full：设置 row 格式的二进制日志记录行的完整信息，包括列值、注释等。这有助于提高应用程序的可恢复性。
- expire_logs_days = 10：设置自动清理过期二进制日志文件的天数。在这个例子中，设置为 10 天。
- gtid_mode = on：启用全局事务 id 模式，这使得基于 gtid 的复制成为可能。
- enforce_gtid_consistency = on：强制执行 gtid 一致性，确保事务在不同的 mysql 实例之间保持一致。
2.如果在windows使用小皮
在小皮面板里设置，如图：
打开bin日志开关
搭建flink cdc java环境
添加maven相关pom
在pom里添加相关flink cdc依赖
<!-- flink connector 基础包-->
<dependency>
<groupid>org.apache.flink</groupid>
<artifactid>flink-connector-base</artifactid>
<version>1.14.4</version>
</dependency>
<!-- cdc mysql 源-->
<dependency>
<groupid>com.ververica</groupid>
<artifactid>flink-sql-connector-mysql-cdc</artifactid>
<version>2.3.0</version>
</dependency>
<!-- flink steam流处理-->
<dependency>
<groupid>org.apache.flink</groupid>
<artifactid>flink-streaming-java_2.11</artifactid>
<version>1.14.4</version>
</dependency>
<!-- flink java客户端-->
<dependency>
<groupid>org.apache.flink</groupid>
<artifactid>flink-clients_2.12</artifactid>
<version>1.14.4</version>
</dependency>
<!-- 开启webui支持，默认是8081，默认没有开启-->
<dependency>
<groupid>org.apache.flink</groupid>
<artifactid>flink-runtime-web_2.12</artifactid>
<version>1.14.4</version>
</dependency>
<!-- flink table api和sql api使得在flink中进行数据处理变得更加简单和高效
通过使用table api和sql api，可以像使用传统的关系型数据库一样，通过编写sql语句或者使用类似于
java的api进行数据处理和分析-->
<dependency>
<groupid>org.apache.flink</groupid>
<artifactid>flink-table-runtime_2.11</artifactid>
<version>1.14.4</version>
</dependency>
<dependency>
<groupid>ch.qos.logback</groupid>
<artifactid>logback-classic</artifactid>
<version>1.2.11</version>
</dependency>
<dependency>
<groupid>org.slf4j</groupid>
<artifactid>slf4j-api</artifactid>
<version>2.0.6</version>
</dependency>
这是一段 maven 依赖配置，它引入了 flink connector base、cdc mysql source、flink streaming java、flink java client、flink runtime web、flink table runtime 和 logback classic。
这些依赖库提供了以下功能：
- flink connector base:flink 的连接器基础包，用于将 flink 与其他系统进行集成。
- cdc mysql source:flink 的 mysql cdc 源，用于从 mysql 数据库中读取数据流。
- flink streaming java:flink 的 java 流处理 api,用于编写并发程序以处理数据流。
- flink java client:flink 的 java api,用于在 java 应用程序中使用 flink。
- flink runtime web:flink 的 web ui,用于监控和管理 flink 集群。
- flink table runtime:flink 的 table api,使在 flink 中进行数据处理变得更加简单和高效。
- logback classic:日志记录库，用于记录应用程序的日志信息。
构建sink
flink cdc(change data capture)中的sink用于将cdc接收到的数据写入外部系统(如数据库或文件系统),以实现数据同步和数据备份等功能，并将其转换为datastream流。然后，sink将这个datastream流写入到外部系统中，以便进行后续的数据处理和分析。
import org.apache.flink.streaming.api.functions.sink.richsinkfunction;
public class customsink extends richsinkfunction<string> {
@override
public void invoke(string value, context context) throws exception {
system.out.println("json->: "+value);
}
}
这段代码定义了一个名为customsink的类，它继承自richsinkfunction类。richsinkfunction是flink cdc中用于将数据写入外部系统(如数据库或文件系统)的函数接口。customsink的作用是将cdc接收到的数据写入外部系统中。具体实现方式由子类customsink来定义。由于这个类继承了richsinkfunction,因此可以使用flink中的其他richsink函数特性，例如设置日志级别、配置连接等,invoke则是处理函数。
main配置运行
如下面的代码，构建flink cdc连接
public static void main(string[] args) throws exception {
mysqlsourcebuilder<string> builder = mysqlsource.builder();
mysqlsource<string> source = builder.hostname("192.168.2.6")
.port(3306)
.databaselist("mydb")
.tablelist("mydb.user")
.username("root")
.password("root")
.deserializer(new jsondebeziumdeserializationschema())
.includeschemachanges(true)
.build();
// 启动webui，绑定本地web-ui端口号
configuration configuration=new configuration();
configuration.setinteger(restoptions.port,8081);
streamexecutionenvironment env = streamexecutionenvironment.getexecutionenvironment(configuration);
env.enablecheckpointing(5000);
env.fromsource(source, watermarkstrategy.nowatermarks(),"mysql source")
.addsink(new customsink());
env.execute();
}
这段代码是使用flink构建一个数据流处理任务，从mysql数据库中读取数据并进行处理。
首先，使用mysqlsourcebuilder创建一个mysqlsource对象，并设置连接参数(hostname、port、databaselist、tablelist、username和password)以及反序列化器(jsondebeziumdeserializationschema)。然后，创建configuration对象并设置webui端口号(restoptions.port),接着使用streamexecutionenvironment创建一个执行环境，启用检查点(checkpointing)并将mysqlsource和自定义的sink添加到执行环境中。最后，执行整个任务。
操作数据库查看结果
如图所示：
{
"before": null,
"after": {
"id": "1661935564737286146",
"qu\_type": 4,
"level": 1,
"image": "",
"content": "dos查看日期、时间",
"create\_time": 1685100091000,
"update\_time": 1685100091000,
"remark": "",
"analysis": ""
},
"source": {
"version": "1.6.4.final",
"connector": "mysql",
"name": "mysql\_binlog\_source",
"ts\_ms": 0,
"snapshot": "false",
"db": "mydb",
"sequence": null,
"table": "t\_user",
"server\_id": 0,
"gtid": null,
"file": "",
**网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**
**需要这份系统化的资料的朋友，可以添加v获取：vip204888 （备注大数据）**
**一个人可以走的很快，但一群人才能走的更远！不论你是正从事it行业的老鸟或是对it行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**
大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**
**需要这份系统化的资料的朋友，可以添加v获取：vip204888 （备注大数据）**
[外链图片转存中...(img-agbmrrp4-1713171905829)]
**一个人可以走的很快，但一群人才能走的更远！不论你是正从事it行业的老鸟或是对it行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**
发表评论