CENTO OS上的网络安全工具（二十三）VSCODE SPARK 容器式编程环境构建

   在vscode上使用maven构建spark的scala编程环境，很大程度上需要不断地从网络上下载各种依赖和插件，而且这一过程复杂而不可控。下面这段，是整个安装过程中/root目录下不断增加的内容。

[root@d7ff8f448a0d /]# cd /root
[root@d7ff8f448a0d ~]# ls -a
.  ..  .bash_logout  .bash_profile  .bashrc  .cshrc  .ssh  .tcshrc  .vscode-server  anaconda-ks.cfg  hadoop  init-vscode.sh  m2repo  maven  spark
[root@d7ff8f448a0d ~]# ls -a
.   .bash_history  .bash_profile  .cshrc  .ssh     .vscode-server   hadoop          m2repo  program
..  .bash_logout   .bashrc        .pki    .tcshrc  anaconda-ks.cfg  init-vscode.sh  maven   spark
[root@d7ff8f448a0d ~]# ls -a
.   .bash_history  .bash_profile  .cache  .dotnet  .ssh     .vscode-server   hadoop          m2repo  program
..  .bash_logout   .bashrc        .cshrc  .pki     .tcshrc  anaconda-ks.cfg  init-vscode.sh  maven   spark
[root@d7ff8f448a0d ~]# ls -a
.   .bash_history  .bash_profile  .cache  .dotnet  .ssh     .vscode-server   hadoop          m2repo  program
..  .bash_logout   .bashrc        .cshrc  .pki     .tcshrc  anaconda-ks.cfg  init-vscode.sh  maven   spark
[root@d7ff8f448a0d ~]# ls -a
.   .bash_history  .bash_profile  .bloop  .config  .dotnet  .pki     .ssh     .vscode-server   hadoop          m2repo  program
..  .bash_logout   .bashrc        .cache  .cshrc   .m2      .redhat  .tcshrc  anaconda-ks.cfg  init-vscode.sh  maven   spark
[root@d7ff8f448a0d ~]# ls -a
.   .bash_history  .bash_profile  .bloop  .config  .dotnet  .m2   .redhat  .tcshrc         anaconda-ks.cfg  init-vscode.sh  maven    spark
..  .bash_logout   .bashrc        .cache  .cshrc   .local   .pki  .ssh     .vscode-server  hadoop           m2repo          program
[root@d7ff8f448a0d ~]#

    另一方面，vscode的离线安装也需要匹配容器中code与外部code的版本。所以完全构造一个不依靠网络随时可启动、还适配任何主机vscode的scala环境着实困难，甚至直接从构建完成的容器导出的镜像，都很难保证再次导入后可用。所以最终我还是放弃了基于Dockerfile构建离线可用的容器环境，转为在所有环境构建完成后，通过export继而import导入最终的镜像。

一、编译容器的Dockerfile

    先准备一个安装了各种语言开发包的容器，以便以后使用vscode连接。当然，为了能够进行hadoop和spark编程，我们把这两个的单机安装过程也跑一遍。装上ssh，这样就可以从vscode连接了。

FROM centos:centos7

#口令参数需要从外部传入
ARG password

#构造更改了清华镜像源的centos7镜像
RUN sed -e 's|^mirrorlist=|#mirrorlist=|g' \
        -e 's|^#baseurl=http://mirror.centos.org/centos|baseurl=https://mirrors.tuna.tsinghua.edu.cn/centos|g' \
        -i.bak \
        /etc/yum.repos.d/CentOS-*.repo\
 && yum clean all

ADD hadoop-3.3.5.tar.gz /root
ADD spark-3.4.0-bin-hadoop3.tgz /root

#拷贝SSH免密登录的相关密钥文件，目前只放置了15个
COPY .ssh /root/.ssh
#拷贝所有待安装软件
COPY ./rpm /root/rpm/.
#拷贝初始化脚本
COPY ./init-vscode.sh /root/.

#构造可swarm一健部署的SSH免密登录镜像-------------------------------------------------------------
#设置初始化脚本的可执行属性
#更改私钥文件属性为仅root用户可见，否则ssh会拒绝执行
#需要为root用户设置口令，否则免密登录还会弹出框，docker镜像默认没有密钥
#安装openssh，并生成服务端密钥，更改强制指纹验证配置项为no，以免弹出指纹询问框
RUN    chmod 0400 /root/.ssh/id_rsa \
 &&    chmod 0600 /root/.ssh/authorized_keys \
 &&    echo ${password} | passwd --stdin root \
 &&    yum localinstall /root/rpm/*.rpm -y \
 &&    /sbin/sshd-keygen \
 &&    echo -e '\nHost *\nStrictHostKeyChecking no\nUserKnownHostsFile=/dev/null' >> etc/ssh/ssh_config \
 &&    rm /root/rpm -rf \
 &&    chmod +x /root/init-vscode.sh \
 &&    mv /root/hadoop-3.3.5 /root/hadoop \
 &&    echo -e "export HADOOP_HOME=/root/hadoop\nexport PATH=\$PATH:\$HADOOP_HOME/bin\n" >> /root/.bashrc\
 &&    mv /root/spark-3.4.0-bin-hadoop3 /root/spark\
 &&    echo -e "export SPARK_HOME=/root/spark \nexport PATH=\$PATH:\$SPARK_HOME/bin">>/root/.bashrc\
 &&    echo -e "export JAVA_HOME=/usr/java/latest">>/root/.bashrc \
 &&    echo -e "export SCALA_HOME=/usr/share/scala">>/root/.bashrc
#默认启动脚本
CMD ["/root/init-vscode.sh"]

    所以初始化脚本的主要作用是把sshd启动起来：

#! /bin/bash

source ~/.bashrc

/sbin/sshd -D &

tail -f /dev/null

二、远程VSCODE安装

    在windows本地安装vscode之类的啥就不提了，直接从安装好了开始：

1. 远程SSH连接容器

    只需要在创建ssh连接的时候填入 ssh root@pighost1 -p9025即可

    仅仅是比原先多了一个端口设定，以便连接到容器绑定的端口上。

2. 插件安装

    安装插件主要是2个大包。

（1）Extension Pack for java

（2）Extension Pack for scala

    不过这个包会捆绑一些我们当下其实并不需要的一些东西，而且在没有注册的时候还经常报错，很是烦人，所以我们也可以选择手动安装必要的插件，主要包括：

** Scala（Metals）**

** Scala Syntax（official）**

** Scala Language Server **

** Scala Snippets**

     亲测这4个装完完全够scala开发了。

3. Metals环境安装

    此时点击Metals图标，会开始安装，等待就好

    里里外外会下载还几个G的东西，所以等待时间会比较长。下载完成后会警告说没有编译工具对象，大意就是指maven的那个pom文件。没关系，还没构建呢。

4. 使用Mave创建一个Scala工程

    构建scalameta/maven-scala-seed.g8，

    点击program（之前我们建的文件夹），再点击ok后，会要求给一个名字，默认是maven-scala-seed。

    然后安静等待，视网络情况，可能得一会儿生成器才会开始工作…这个“一会儿”可能长达一分钟，所以一定要选择** 相信 **…注意看网络速率那里（如果有装相关的监视软件或者杀毒软件小插件之类的话）其实一直在下载，在没有下载完成之前，这个自动生成的操作是不会开始的。

    生成结束后，vscode会询问是否在新窗口中打开项目，选择yes（这样项目会跳到pom文件所在的目录下，从而不会导致生成可执行文件时遭遇no target错误）。

关闭原来的窗口，使用新打开的窗口就好。此时项目的目录结构应该时这样的，可以看到pom文件已经出现了：

    这时一般vscode会很贴心的问要不要import build一下。**此时千万不要import！！！此时千万不要import！！！此时千万不要import！！！**

     否则可能会由于java版本不匹配造成某些地方的配置错误，我也不知道错在哪（估计是在metals的java环境同步设置中，懒得找了）。pom文件中默认配置jdk是1.8版本的。由于我们使用的是jdk 11，所以pom文件的这里需要改动一下：

     1即可：

     然后这次VS从的仍然会询问是否import，这次就可以选择import build了

import过程中，maven会运行bloopinstall，这个也需要花很长时间。

    最后在输出窗口中会看到SUCCESS。但是不要以为这就是结束。

    然后就是需要继续观察下载速率，等待下载……总之装这个一路都很玄学，因为有些下载在输出窗口里面是能看到的（如果选择了观察logs），有些下载操作在窗口是什么都看不到的——如果你以为什么动静都看不到就是装完了而试图区执行代码的时候，一般会收到internal error。

    **信则有，不信就没有。**我们要坚定地相信vscode会帮助我们搞定一切的（=反正不信也没有别的什么办法……）。直到所有配置工作搞定之后，应该能发现示例代码可执行的类上方出现了“run | debug”这样的图标，代表vscode准备好了。

     如果怎么都等不到这个结果，不妨直接运行一下代码，当然会收到Internal error的错误。不过这次运行似乎会刺激metals继续工作。因为不久后右下角会出现indexing以及scalafix工作的字样，完成后一般metals的packages窗口内会出现内容，而不是那两个找不到scala程序及要求创建的按钮：

     如果是jdk11的环境（我也不知道1.8会不会有类似问题……因为下面这一步我没有配置的时候也能成功，但有时候不配就是不成功：）

    在帮助栏中执行run doctor：

    得到如下界面：

错误的图木有了，正确的图示如上，总之这个semanticdb插件没有的话，就不行，还得手动安装一下。——到maven仓库里面找到，在pom文件中添加如下即可：

    <!-- https://mvnrepository.com/artifact/org.scalameta/semanticdb -->
    <dependency>
        <groupId>org.scalameta</groupId>
        <artifactId>semanticdb_2.12</artifactId>
        <version>4.1.6</version>
    </dependency>

     此时执行程序，应该结果就不会有问题了：

5. 导入Spark依赖

     对于Spark，需要向pom文件中导入对应的依赖，如上。

    最终的pom文件：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>testspark</artifactId>
  <version>1.0-SNAPSHOT</version>
  <name>testspark</name>
  <description>A minimal Scala project using the Maven build tool.</description>
  <licenses>
    <license>
      <name>My License</name>
      <url>http://....</url>
      <distribution>repo</distribution>
    </license>
  </licenses>

  <properties>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
    <encoding>UTF-8</encoding>
    <scala.version>2.13.11</scala.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>2.13.11</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.scalameta/semanticdb -->
    <dependency>
        <groupId>org.scalameta</groupId>
        <artifactId>semanticdb_2.12</artifactId>
        <version>4.1.6</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.13</artifactId>
        <version>3.4.0</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.13</artifactId>
        <version>3.4.0</version>
        <scope>provided</scope>
    </dependency>

    <!-- Test -->
    <dependency>
      <groupId>org.scalameta</groupId>
      <artifactId>munit_2.13</artifactId>
      <version>0.7.29</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <sourceDirectory>src/main/scala</sourceDirectory>
    <testSourceDirectory>src/test/scala</testSourceDirectory>
    <plugins>
      <plugin>
        <!-- see http://davidb.github.com/scala-maven-plugin -->
        <groupId>net.alchim31.maven</groupId>
        <artifactId>scala-maven-plugin</artifactId>
        <version>3.3.2</version>
        <configuration></configuration>
        <executions>
          <execution>
            <goals>
              <goal>compile</goal>
              <goal>testCompile</goal>
            </goals>
            <configuration>
              <args></args>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

    然后就是继续等待vscode下载。当然，一般来说这些java包下载存储在 ~/.m2下。我们也可以直接从mave repository上下载java包放到对应目录下即可。

    放上经典的分词并且计数代码，执行：

     结果可以到容器里面观察一下：

6. Scala 和 Java 混合编程

     Scala本身就是从Java生长出来的，所以在scala环境中使用java代码，或者在java中使用scala代码应该说是很直接的需求。要满足这样混合编程的条件，需要在pom文件的plugin中添加插件：

      <!--为混合编程添加-->
      <plugin>
        <groupId>org.codehaus.mojo</groupId>
        <artifactId>build-helper-maven-plugin</artifactId>
        <executions>
          <execution>
            <id>add-source</id>
            <phase>generate-sources</phase>
            <goals>
              <goal>add-source</goal>
            </goals>
            <configuration>
              <sources>
                <source>src/main/scala</source>
              </sources>
            </configuration>
          </execution>
          <execution>
            <id>add-test-source</id>
            <phase>generate-sources</phase>
            <goals>
              <goal>add-test-source</goal>
            </goals>
            <configuration>
              <sources>
                <source>src/test/scala</source>
              </sources>
            </configuration>
          </execution>
        </executions>
      </plugin>

    然后直接在工作目录中增加java文件：

     就可以在scala中调用了：

     结果如下：       ![](https://img-blog.csdnimg.cn/4c7662b5b1c34941bfa194be0fe33bfc.png)

三、导出镜像

1. 使用docker export导出当前的容器

[root@pighost share]# docker ps -a
CONTAINER ID   IMAGE        COMMAND                  CREATED         STATUS         PORTS                                   NAMES
76d0b875683b   pig/vscode   "/root/init-vscode.sh"   8 minutes ago   Up 8 minutes   0.0.0.0:9025->22/tcp, :::9025->22/tcp   pig
[root@pighost share]# docker export -o pigvscode.tar pig

2. 使用dockers import构建镜像

[root@pighost share]# docker import pigvscode.tar pig/vscode:scala
sha256:ed572d045eba3c481dc557b2646958548a3e6f469ece866c5605aaef3cdccdc3
[root@pighost share]# docker images
REPOSITORY   TAG       IMAGE ID       CREATED         SIZE
pig/vscode   scala     ed572d045eba   9 seconds ago   4.49GB
centos       centos7   eeb6ee3f44bd   21 months ago   204MB

3. 基于镜像启动容器

    由于我们是直接使用容器构建的镜像，所以初始化脚本不会在容器载入时执行了，需要我们在载入过程中显示指定。（当然，这样我们以后就无法在swarm中使用了。如果需要，应该基于该镜像，重新使用Dockerfile构建一个新的镜像并指明CMD就好了。）

[root@pighost share]# docker run -itd --name pig -p 9025:22 pig/vscode:scala /root/init-vscode.sh
37b6651fe96fe58ae17eda584310e83778a6d3d012ba55584a53a42c1ecd7065
[root@pighost share]#

     然后启动vscode直接进行远程连接就可以了。

四、Spark-shell交互式分析

    使用scala编程，有很多事可做。当然，在开始编程之前，使用spark-shell交互式环境，体验一下scala的使用效果，也是不错的。直接在命令行中执行spark-shell就可以了：

1. 读csv文件：

    val df = spark.read.option("header",true).csv("/sample/DOS/*.csv")

    或

    val df = spark.read.option("header",true).format("csv").load("/sample/DOS/*.csv")

scala> val df = spark.read.option("header",true).csv("/sample/DOS/*.csv")
df: org.apache.spark.sql.DataFrame = [sid: string, message_type: string ... 23 more fields]

2. 查看读入文件的统计信息

    df.count

    df.columns

scala> df.count
res0: Long = 155395                                                             

scala> df.columns
res1: Array[String] = Array(sid, message_type, attack_type, src_ip, src_port, src_mac, src_country, src_province, src_city, src_location, dest_ip, dest_port, dest_mac, dest_country, dest_province, dest_city, dest_location, device_id, application_protocol, protocol, ip_version, timestamp, risk_level, length, message)

3. 查看读入数据的模式信息

    df.schema

scala> df.schema
res2: org.apache.spark.sql.types.StructType = StructType(StructField(sid,StringType,true),StructField(message_type,StringType,true),StructField(attack_type,StringType,true),StructField(src_ip,StringType,true),StructField(src_port,StringType,true),StructField(src_mac,StringType,true),StructField(src_country,StringType,true),StructField(src_province,StringType,true),StructField(src_city,StringType,true),StructField(src_location,StringType,true),StructField(dest_ip,StringType,true),StructField(dest_port,StringType,true),StructField(dest_mac,StringType,true),StructField(dest_country,StringType,true),StructField(dest_province,StringType,true),StructField(dest_city,StringType,true),StructField(dest_location,StringType,true),StructField(device_id,StringType,true),Stru...

scala> df.printSchema
root
 |-- sid: string (nullable = true)
 |-- message_type: string (nullable = true)
 |-- attack_type: string (nullable = true)
 |-- src_ip: string (nullable = true)
 |-- src_port: string (nullable = true)
 |-- src_mac: string (nullable = true)
……………………
 |-- device_id: string (nullable = true)
 |-- application_protocol: string (nullable = true)
 |-- protocol: string (nullable = true)
 |-- ip_version: string (nullable = true)
 |-- timestamp: string (nullable = true)
 |-- risk_level: string (nullable = true)
 |-- length: string (nullable = true)
 |-- message: string (nullable = true)

    刚导入进来的全是string格式

4. 列操作

（1）选择列

scala> val smalldf = df.select(col("src_ip"),col("src_port"),col("dest_ip"),col("dest_port"),col("message"))
scala> smalldf.show()
+---------------+--------+---------------+---------+--------------------+
|         src_ip|src_port|        dest_ip|dest_port|             message|
+---------------+--------+---------------+---------+--------------------+
|  192.168.11.20|   38021|172.0.1.68     |      123|FRN DOS Possible ...|
………………………………
|  192.168.11.20|   47430|172.0.1.68     |      123|FRN DOS Possible ...|
+---------------+--------+---------------+---------+--------------------+
only showing top 20 rows

smalldf: Unit = ()

    选择列的简单方式：


scala> lines.select('src_ip).show()
+---------------+                                                               
|         src_ip|
+---------------+
|  192.168.1.46|
|  192.168.1.46|
|  192.168.1.46|
…………………………
only showing top 20 rows

（2）增加列

scala> lines.select('src_ip).withColumn("count",lit(1)).show
+---------------+-----+
|         src_ip|count|
+---------------+-----+
|  192.168.1.111|    1|
|  192.168.1.112|    1|

5. 去重

    去重只有distinct可用，如果是只正对一个字段的话，得把这个字段选出来

scala> lines.count
res9: Long = 155395                                                             

scala> lines.distinct.count
23/05/24 02:01:31 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.
res10: Long = 154022                                                            

scala> lines.select('src_ip).distinct.count
res11: Long = 1454                                                              

scala>

    可以看出效果截然不同

6. 过滤

scala> lines.count
res16: Long = 155395                                                            

scala> lines.filter("dest_port='123'").count
res17: Long = 97147

scala> lines.filter("src_ip='192.168.0.11' and dest_port='123'").count
res18: Long = 3202

    where 就是sql语句where后面得表达式

scala> lines.where('dest_ip.like("192%")).select('src_ip,'src_port,'dest_ip,'dest_port).show
+---------------+--------+---------------+---------+
|         src_ip|src_port|        dest_ip|dest_port|
+---------------+--------+---------------+---------+
|  192.168.1.11 |   38021|    172.18.0.13|      123|

    列语法糖 $ === =!=

scala> lines.filter($"protocol"==="tcp").select('src_ip,'src_port,'dest_ip,'dest_port,'protocol).show
+---------------+--------+--------------+---------+--------+
|         src_ip|src_port|       dest_ip|dest_port|protocol|
+---------------+--------+--------------+---------+--------+
|  192.168.1.112|   61841|    172.17.0.1|     2710|     tcp|

7. 统计

scala> lines.groupBy('dest_ip).count().orderBy('count.desc).show()
+---------------+-----+                                                         
|        dest_ip|count|
+---------------+-----+
|   192.168.1.11|  273|
|   192.168.1.12|  148|
|   192.168.1.13|  142|

8. 类型转换

    spark有3中主要类型，RDD、Dataframe和Dataset，各有各的优势，所以经常需要转换使用。

（1）DataFrame转Dataset

scala> val fivetuples = lines.select('src_ip,'src_port,'dest_ip,'dest_port,'protocol)
fivetuples: org.apache.spark.sql.DataFrame = [src_ip: string, src_port: string ... 3 more fields]

scala> case class FIVETUPLE(src_ip:String,src_port:String,dest_ip:String,dest_port:String,protocol:String)
defined class FIVETUPLE

scala> val fset = fivetuples.as[FIVETUPLE]
fset: org.apache.spark.sql.Dataset[FIVETUPLE] = [src_ip: string, src_port: string ... 3 more fields]

（2）DataSet转DataFrame

scala> val fdf = fset.toDF
fdf: org.apache.spark.sql.DataFrame = [src_ip: string, src_port: string ... 3 more fields]

scala> fdf.columns
res31: Array[String] = Array(src_ip, src_port, dest_ip, dest_port, protocol)

（3）DataFrame转RDD

scala> val frdd = fdf.rdd
frdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[124] at rdd at <console>:23

（4）RDD转DataFrame

    转回来需要map一下

scala> val fdf2 = frdd.map(l=>FIVETUPLE(l(0).toString(),l(1).toString(),l(2).toString(),l(3).toString(),l(4).toString())).toDF
fdf2: org.apache.spark.sql.DataFrame = [src_ip: string, src_port: string ... 3 more fields]

（5）RDD和DataSet互转

scala> val fset2 = frdd.map(l=>FIVETUPLE(l(0).toString(),l(1).toString(),l(2).toString(),l(3).toString(),l(4).toString())).toDS
fset2: org.apache.spark.sql.Dataset[FIVETUPLE] = [src_ip: string, src_port: string ... 3 more fields]

scala> fset2.rdd
res32: org.apache.spark.rdd.RDD[FIVETUPLE] = MapPartitionsRDD[129] at rdd at <console>:24

9. 碰撞

     在交互式环境中进行碰撞操作，可以将需要碰撞的一个集合通过array或者list导出来。比如：

（1）获取array

scala> val ip11 = lines.where('dest_ip.like("%11")).select('dest_ip).collect()
ip11: Array[org.apache.spark.sql.Row] = Array([192.168.………………

（2）获取List

scala> val ip11 = lines.where('dest_ip.like("%11")).select('dest_ip).collectAsList()
ip11: java.util.List[org.apache.spark.sql.Row] = [[192.168.……

（3）转为可用的数组或列表

    但以上两种获得的都是Row类型，在碰撞程序中不好用。要获得程序中好用的List，需要转化为rdd，然后使用map方法将row中的元素提取出来：

scala> val ip11 = lines.where('dest_ip.like("%111")).distinct.select('dest_ip).rdd.map(r=>r(0)).collect()
ip11: Array[Any] = Array(192.168.0.111，………………

    可再使用toList转为List

（4）碰撞发现ip名单内的数据

scala> lines.where('src_ip.isin(ip11:_*)).show

（5）碰撞排除ip名单内的数据

scala> lines.where(!'dest_ip.isin(ip11:_*)).select('dest_ip).show

10. SQL操作

scala> val lines = spark.read.option("header",true).csv("/sample/DOS/*.csv")
lines: org.apache.spark.sql.DataFrame = [sid: string, message_type: string ... 23 more fields]

scala> lines.createOrReplaceTempView("mytable")

scala> spark.sql("select src_ip from mytable").show
+---------------+
|         src_ip|
+---------------+
|    192.168.1.1|

（1）分组统计排序

scala> spark.sql("select dest_ip,dest_port,count(*) as count from mytable group by dest_port,dest_ip order by count desc").show
+---------------+---------+-----+                                               
|        dest_ip|dest_port|count|
+---------------+---------+-----+
|  192.168.1.111|      123|  240|
|  192.168.1.112|      123|  122|

（2）对分组排序条件进行限制

scala> spark.sql("select dest_ip,dest_port,count(*) as count from mytable group by dest_port,dest_ip having count<20 order by count desc").show
+---------------+---------+-----+                                               
|        dest_ip|dest_port|count|
+---------------+---------+-----+
| 192.168.21.222|      111|   19|
| 192.168.21.222|      111|   19|

scala> spark.sql("select dest_ip,dest_port,count(*) as count from mytable where dest_ip like '173%' group by dest_port,dest_ip having count<20 order by count desc").show
+---------------+---------+-----+                                               
|        dest_ip|dest_port|count|
+---------------+---------+-----+
|173.112.111.111|      111|   19|
|173.112.111.112|      111|   19|

（3）嵌套碰撞

scala> spark.sql("select src_ip from mytable where src_ip in (select dest_ip from mytable)").show
+---------------+
|         src_ip|
+---------------+
|       10.0.0.1|

（4）仅显示5行：

scala> spark.sql("select src_ip,count(*) as count from mytable group by src_ip having count>100 limit 5").show
+---------------+-----+
|         src_ip|count|
+---------------+-----+
|    192.168.1.1|  726|
|    192.168.1.2| 2030|
|    192.168.1.3|  328|
|    192.168.1.4|  204|
|    192.168.1.5|  190|

（5）去重

scala> spark.sql("select protocol from mytable").show
+--------+
|protocol|
+--------+
|     udp|
|     udp|
|     udp|
…………
|     udp|
|     udp|
+--------+
only showing top 20 rows

scala> spark.sql("select distinct protocol from mytable").show
+--------+
|protocol|
+--------+
|     tcp|
|     udp|
+--------+

11. DataSet操作

    Dataset操作和Dataframe操作类似：

    过滤 分组聚合 排序

scala> fset.select('src_ip,'dest_ip).groupBy('src_ip).count.orderBy('count.desc).show
+---------------+-----+
|         src_ip|count|
+---------------+-----+
|       10.0.0.1|19720|
|       10.0.0.2| 8037|
|       10.0.0.3| 5778|

    可以map

scala> fset.map(l => (l.src_ip,1)).show
+---------------+---+
|             _1| _2|
+---------------+---+
|  192.168.10.11|  1|
|  192.168.10.12|  1|
|  192.168.10.12|  1|
|  192.168.10.11|  1|

    reduce 统计行数

scala> fset.map(l=>1).reduce(_+_)
res44: Int = 155395

scala> fset.count
res45: Long = 155395

12. RDD操作：

    RDD的最大好处就是可以支持灵活的map-reduce，最重要的事可以支持K-V操作。

    按src ip过滤

scala> frdd.count
res53: Long = 155395

scala> frdd.filter(l => l(0)=="192.168.0.1").count
res54: Long = 3202

    转KV操作

scala> frdd.keyBy(l=>l(0)).first

    countByKey

scala> frdd.keyBy(l=>l(0)).countByKey

标签：大数据 hadoop spark

本文转载自: https://blog.csdn.net/lhyzws/article/details/130785417
版权归原作者 lhyzws 所有，如有侵权，请联系我们删除。