devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

2024年网络安全最新Flink Table API 与 SQL 编程整理，推荐学习

作者：Monodyee | 2024-06-05 07:58:39

赞

踩

2024年网络安全最新Flink Table API 与 SQL 编程整理，推荐学习

如何自学黑客&网络安全

黑客零基础入门学习路线&规划

初级黑客
1、网络安全理论知识（2天）
①了解行业相关背景，前景，确定发展方向。
②学习网络安全相关法律法规。
③网络安全运营的概念。
④等保简介、等保规定、流程和规范。（非常重要）

2、渗透测试基础（一周）
①渗透测试的流程、分类、标准
②信息收集技术：主动/被动信息搜集、Nmap工具、Google Hacking
③漏洞扫描、漏洞利用、原理，利用方法、工具（MSF）、绕过IDS和反病毒侦察
④主机攻防演练：MS17-010、MS08-067、MS10-046、MS12-20等

3、操作系统基础（一周）
①Windows系统常见功能和命令
②Kali Linux系统常见功能和命令
③操作系统安全（系统入侵排查/系统加固基础）

4、计算机网络基础（一周）
①计算机网络基础、协议和架构
②网络通信原理、OSI模型、数据转发流程
③常见协议解析（HTTP、TCP/IP、ARP等）
④网络攻击技术与网络安全防御技术
⑤Web漏洞原理与防御：主动/被动攻击、DDOS攻击、CVE漏洞复现

5、数据库基础操作（2天）
①数据库基础
②SQL语言基础
③数据库安全加固

6、Web渗透（1周）
①HTML、CSS和JavaScript简介
②OWASP Top10
③Web漏洞扫描工具
④Web渗透工具：Nmap、BurpSuite、SQLMap、其他（菜刀、漏扫等）
恭喜你，如果学到这里，你基本可以从事一份网络安全相关的工作，比如渗透测试、Web 渗透、安全服务、安全分析等岗位；如果等保模块学的好，还可以从事等保工程师。薪资区间6k-15k

到此为止，大概1个月的时间。你已经成为了一名“脚本小子”。那么你还想往下探索吗？

如果你想要入坑黑客&网络安全，笔者给大家准备了一份：282G全网最全的网络安全资料包评论区留言即可领取！

7、脚本编程（初级/中级/高级）
在网络安全领域。是否具备编程能力是“脚本小子”和真正黑客的本质区别。在实际的渗透测试过程中，面对复杂多变的网络环境，当常用工具不能满足实际需求的时候，往往需要对现有工具进行扩展，或者编写符合我们要求的工具、自动化脚本，这个时候就需要具备一定的编程能力。在分秒必争的CTF竞赛中，想要高效地使用自制的脚本工具来实现各种目的，更是需要拥有编程能力.

如果你零基础入门，笔者建议选择脚本语言Python/PHP/Go/Java中的一种，对常用库进行编程学习；搭建开发环境和选择IDE,PHP环境推荐Wamp和XAMPP， IDE强烈推荐Sublime；·Python编程学习，学习内容包含：语法、正则、文件、 网络、多线程等常用库，推荐《Python核心编程》，不要看完；·用Python编写漏洞的exp,然后写一个简单的网络爬虫；·PHP基本语法学习并书写一个简单的博客系统；熟悉MVC架构，并试着学习一个PHP框架或者Python框架 (可选)；·了解Bootstrap的布局或者CSS。

8、超级黑客
这部分内容对零基础的同学来说还比较遥远，就不展开细说了，附上学习路线。
img

网络安全工程师企业级学习路线

img
如图片过大被平台压缩导致看不清的话，评论区点赞和评论区留言获取吧。我都会回复的

视频配套资料&国内外网安书籍、文档&工具

当然除了有配套的视频，同时也为大家整理了各种文档和书籍资料&工具，并且已经帮大家分好类了。

img
一些笔者自己买的、其他平台白嫖不到的视频教程。
img

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化资料的朋友，可以点击这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

SQL： WordCount案例

SELECT word,COUNT(\*) AS cnt FROM MyTable GROUP BY word


  • 1
  • 2

【1】声明式： 用户只关系做什么，不用关心怎么做；
【2】高性能： 支持查询优化，可以获取更好的执行性能，因为它的底层有一个优化器，跟SQL底层有优化器是一样的。
【3】流批统一： 相同的统计逻辑，即可以流模型运行，也可以批模式运行；
【4】标准稳定： 语义遵循SQL标准，不易改动。当升级等底层修改，不用考虑API兼容问题；
【5】易理解： 语义明确，所见即所得；

Table API 特点

Table API使得多声明的数据处理写起来比较容易。

1 #例如，我们将a<10的数据过滤插入到xxx表中
2 table.filter(a<10).insertInto("xxx")
3 #我们将a>10的数据过滤插入到yyy表中
4 table.filter(a>10).insertInto("yyy")


  • 1
  • 2
  • 3
  • 4
  • 5

Talbe是Flink自身的一种API使得更容易扩展标准的SQL（当且仅当需要的时候），两者的关系如下：
在这里插入图片描述

Table API 编程

WordCount编程示例

package org.apache.flink.table.api.example.stream;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.OldCsv;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

public class JavaStreamWordCount {

    public static void main(String[] args) throws Exception {
                //获取执行环境：CTRL + ALT + V
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
                //指定一个路径
        String path = JavaStreamWordCount.class.getClassLoader().getResource("words.txt").getPath();
                //指定文件格式和分隔符，对应的Schema(架构）这里只有一列，类型是String
        tEnv.connect(new FileSystem().path(path))
            .withFormat(new OldCsv().field("word", Types.STRING).lineDelimiter("\n"))
            .withSchema(new Schema().field("word", Types.STRING))
            .inAppendMode()
            .registerTableSource("fileSource");//将source注册到env中
                //通过 scan 拿到table，然后执行table的操作。
        Table result = tEnv.scan("fileSource")
            .groupBy("word")
            .select("word, count(1) as count");
                //将table输出
        tEnv.toRetractStream(result, Row.class).print();
                //执行
        env.execute();
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

怎么定义一个 Table

Table myTable = tableEnvironment.scan("myTable") 都是从Environment中scan出来的。而这个myTable 又是我们注册进去的。问题就是有哪些方式可以注册Table。
【1】Table descriptor： 类似于上述的WordCount，指定一个文件系统fs，也可以是kafka等，还需要一些格式和Schema等。

tEnv.connect(new FileSystem().path(path))
            .withFormat(new OldCsv().field("word", Types.STRING).lineDelimiter("\n"))
            .withSchema(new Schema().field("word", Types.STRING))
            .inAppendMode()
            .registerTableSource("fileSource");//将source注册到env中


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

【2】自定义一个 Table source： 然后把自己的Table source注册进去。

TableSource csvSource = new CsvTableSource(path,new String[]{"word"},new TypeInformation[]{Types.STRING});
tEnv.registerTableSource("sourceTable2", csvSource);


  • 1
  • 2
  • 3

【3】注册一个 DataStream： 例如下面一个String类型的DataStream，命名为myTable3对应的schema只有一列叫word。

DataStream<String> stream = ...
// register the DataStream as table " myTable3" with 
// fields "word"
tableEnv.registerDataStream("myTable3", stream, "word");


  • 1
  • 2
  • 3
  • 4
  • 5

动态表

如果流中的数据类型是case class可以直接根据case class的结构生成table

tableEnv.fromDataStream(ecommerceLogDstream)


  • 1
  • 2

或者根据字段顺序单独命名：用单引放到字段前面来标识字段名。

tableEnv.fromDataStream(ecommerceLogDstream,'mid,'uid ......)


  • 1
  • 2

最后的动态表可以转换为流进行输出，如果不是简单的插入就使用toRetractStream

table.toAppendStream[(String,String)]


  • 1
  • 2

如何输出一个table

当我们获取到一个结构表的时候（table类型）执行insertInto目标表中：resultTable.insertInto("TargetTable");

【1】Table descriptor： 类似于注入，最终使用Sink进行输出，例如如下输出到targetTable中，主要是最后一段的区别。

tEnv
.connect(new FileSystem().path(path)).withFormat(new OldCsv().field("word", Types.STRING)
.lineDelimiter("\n")).withSchema(new Schema()
.field("word", Types.STRING))
.registerTableSink("targetTable");


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

【2】自定义一个 Table sink： 输出到自己的 sinkTable2注册进去。

TableSink csvSink = new CsvTableSink(path,new String[]{"word"},new TypeInformation[]{Types.STRING});
tEnv.registerTableSink("sinkTable2", csvSink);


  • 1
  • 2
  • 3

【3】输出一个 DataStream： 例如下面产生一个RetractStream，对应要给Tuple2的联系。Boolean这行记录时add还是delete。如果使用了groupby，table 转化为流的时候只能使用toRetractStream。得到的第一个boolean型字段标识 true就是最新的数据(Insert)，false表示过期老数据(Delete)。如果使用的api包括时间窗口，那么窗口的字段必须出现在groupBy中。

// emit the result table to a DataStream
DataStream<Tuple2<Boolean, Row>> stream = tableEnv.toRetractStream(resultTable, Row.class)
stream.filter(_._1).print()


  • 1
  • 2
  • 3
  • 4

案例代码：

package com.zzx.flink

import java.util.Properties

import com.alibaba.fastjson.JSON
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.table.api.java.Tumble
import org.apache.flink.table.api.{StreamTableEnvironment, Table, TableEnvironment}



object FlinkTableAndSql {
  def main(args: Array[String]): Unit = {
    //执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置 时间特定为 EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取数据 MyKafkaConsumer 为自定义的 kafka 工具类,并传入 topic
    val dstream: DataStream[String] = env.addSource(MyKafkaConsumer.getConsumer("FLINKTABLE&SQL"))

    //将字符串转换为对象
    val ecommerceLogDstream:DataStream[SensorReding] = dstream.map{
     /\* 引入如下依赖
 <dependency>
 <groupId>com.alibaba</groupId>
 <artifactId>fastjson</artifactId>
 <version>1.2.36</version>
 </dependency>\*/
          //将 String 转换为 SensorReding
      jsonString => JSON.parseObject(jsonString,classOf[SensorReding])
    }

    //告知 watermark 和 evetTime如何提取
    val ecommerceLogWithEventTimeDStream: DataStream[SensorReding] =ecommerceLogDstream.assignTimestampsAndWatermarks(
          new BoundedOutOfOrdernessTimestampExtractor[SensorReding](Time.seconds(0)) {
      override def extractTimestamp(t: SensorReding): Long = {
        t.timestamp
      }
    })
    //设置并行度
    ecommerceLogDstream.setParallelism(1)

    //创建 Table 执行环境
    val tableEnv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(env)
    var ecommerceTable: Table = tableEnv.fromTableSource(ecommerceLogWithEventTimeDStream ,'mid,'uid,'ch,'ts.rowtime)

    //通过 table api进行操作
    //每10秒统计一次各个渠道的个数 table api解决
    //groupby window=滚动式窗口 用 eventtime 来确定开窗时间
    val resultTalbe: Table = ecommerceTable.window(Tumble over 10000.millis on 'ts as 'tt).groupBy('ch,'tt).select('ch,'ch.count)
    var ecommerceTalbe: String = "xxx"
    //通过 SQL 执行
    val resultSQLTable: Table = tableEnv.sqlQuery("select ch,count(ch) from "+ ecommerceTalbe +"group by ch,Tumble(ts,interval '10' SECOND")

    //把 Table 转化成流输出
    //val appstoreDStream: DataStream[(String,String,Long)] = appstoreTable.toAppendStream[(String,String,Long)]
    val resultDStream: DataStream[(Boolean,(String,Long))] = resultSQLTable.toRetractStream[(String,Long)]
    //过滤
    resultDStream.filter(_._1)
    env.execute()
  }
}
object MyKafkaConsumer {
  def getConsumer(sourceTopic: String): FlinkKafkaConsumer011[String] ={
  val bootstrapServers = "hadoop1:9092"
  // kafkaConsumer 需要的配置参数
  val props = new Properties
  // 定义kakfa 服务的地址，不需要将所有broker指定上
  props.put("bootstrap.servers", bootstrapServers)
  // 制定consumer group
  props.put("group.id", "test")

  // 是否自动确认offset
  props.put("enable.auto.commit", "true")
  // 自动确认offset的时间间隔
  props.put("auto.commit.interval.ms", "1000")
  // key的序列化类
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  // value的序列化类
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  //从kafka读取数据,需要实现 SourceFunction 他给我们提供了一个
  val consumer = new FlinkKafkaConsumer011[String](sourceTopic, new SimpleStringSchema, props)
  consumer
  }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92

关于时间窗口

【1】用到时间窗口，必须提前声明时间字段，如果是processTime直接在创建动态表时进行追加就可以。如下的ps.proctime。

val ecommerceLogTable: Table = tableEnv
    .fromDataStream( ecommerceLogWithEtDstream,
        `mid,`uid,`appid,`area,`os,`ps.proctime )


  • 1
  • 2
  • 3
  • 4

【2】如果是EventTime要在创建动态表时声明。如下的ts.rowtime。

val ecommerceLogTable: Table = tableEnv
    .fromDataStream( ecommerceLogWithEtDstream,
        'mid,'uid,'appid,'area,'os,'ts.rowtime)


  • 1
  • 2
  • 3
  • 4

【3】滚动窗口可以使用Tumble over 10000.millis on来表示




还有兄弟不知道网络安全面试可以提前刷题吗？费时一周整理的160+网络安全面试题，金九银十，做网络安全面试里的显眼包！


王岚嵚工程师面试题（附答案），只能帮兄弟们到这儿了！如果你能答对70%，找一个安全工作，问题不大。


对于有1-3年工作经验，想要跳槽的朋友来说，也是很好的温习资料！


【完整版领取方式在文末！！】


***93道网络安全面试题***


![](https://img-blog.csdnimg.cn/img_convert/6679c89ccd849f9504c48bb02882ef8d.png)








![](https://img-blog.csdnimg.cn/img_convert/07ce1a919614bde78921fb2f8ddf0c2f.png)





![](https://img-blog.csdnimg.cn/img_convert/44238619c3ba2d672b5b8dc4a529b01d.png)





内容实在太多，不一一截图了


### 黑客学习资源推荐


最后给大家分享一份全套的网络安全学习资料，给那些想学习 网络安全的小伙伴们一点帮助！


对于从来没有接触过网络安全的同学，我们帮你准备了详细的学习成长路线图。可以说是最科学最系统的学习路线，大家跟着这个大的方向学习准没问题。



声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/675533
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。