PkpmSpark

大数据分析 三维数据挖掘运算分析程序

机理

获取mongodb中存储的三维数据进行分析,归并,聚合,加权运算，最后计算完成后的结果存入redis

环境配置

1.mongodb中需含有名称为mydb数据库与名为netflows的collection

2.需要配置kafka 并新建topic名称为order(若不需要由kafka做触发源则不用配置)

3.需配置redis，无用户名密码登录

4.需spark运行环境 2.4.4测试通过

5.需scala sdk 2.11.12测试通过

程序配置

1.kafka消费实例配置在OrderTopicKafka类中，请跟据需要修改，可由工厂模式附加新构造上去

2.redis链接实例配置在RedisConnector类中，请跟据需要修改

3.mongodb链接实例配置在SparkMainTask的main函数中，请跟据需要修改

编译

本项目配置了maven编译，在idea命令行执行mvn assembly:assembly即可编译生成

数据源

本项目数据源来自mongodb，数据采集可以传入kafka借由如下链接项目完成kafka向mongodb的同步，采集端只需保证json格式：

https://github.com/MethodJiao/Kafka2Mongodb

也可直接在mongodb中仿造执行如下insert语句制造数据，数据格式如下：

1.ChildNode中数组可嵌入多个立方体以此描述三维空间

2.HighPt，LowPt分别为立方体体的对角线端点两点

3.Name可以存储当前立方体表述对象名

4.Origin为当前立方体的原点定位

5.YPRangle为立方体姿态角

db.getCollection("netflows").insert( {
    _id: ObjectId("5dededecb3e6784f020c4e90"),
    RootNode: {
        ChildNode: [
            {
                HighPt: {
                    x: 83956,
                    y: 76703,
                    z: 2900
                },
                IsSet: {
                    set: false
                },
                LowPt: {
                    x: 80155,
                    y: 76502,
                    z: 2500
                },
                Name: {
                    name: "PBStructBeam"
                },
                Origin: {
                    x: 83955.9781857537,
                    y: 76602.633138063,
                    z: 2900
                },
                YPRangle: {
                    pitch: 0,
                    roll: 90,
                    yaw: -90
                }
            },
            {
                HighPt: {
                    x: 83557,
                    y: 74903,
                    z: 2900
                },
                IsSet: {
                    set: false
                },
                LowPt: {
                    x: 80756,
                    y: 74702,
                    z: 2420
                },
                Name: {
                    name: "PBStructBeam"
                },
                Origin: {
                    x: 83556.179,
                    y: 74802.2127855456,
                    z: 2900
                },
                YPRangle: {
                    pitch: 0,
                    roll: 90,
                    yaw: -90
                }
            }
        ],
        KeyValue: NumberInt("-6567")
    }
} );

运行

示例：

bin/spark-submit --class spark.bim.SparkMainTask --master spark://10.100.140.35:7077 /Users/method.jiao/code/pkpmspark/target/pkpmspark-1.0-SNAPSHOT-jar-with-dependencies.jar

参数修改：

1.配置ip 10.100.140.35:7077

2.jar包编译路径 /Users/method.jiao/code/pkpmspark/target/pkpmspark-1.0-SNAPSHOT-jar-with-dependencies.jar

PS

1.如果想在idea直接本地启动不上传task到spark集群的话需要在sparksession加个.master("local")如：

val sparkSession = SparkSession.builder()
  .appName("PKPMBimAnalyse")
  .config("spark.mongodb.input.uri", "mongodb://10.100.140.35/mydb.netflows")
  .master("local")
  .getOrCreate()

2.如果上传task到集群务必去掉master属性

3.如果想在idea远程提交并调试 如：

val sparkSession = SparkSession.builder()
  .appName("PKPMBimAnalyse")
  .config("spark.mongodb.input.uri", "mongodb://10.100.140.35/mydb.netflows")
  .master("spark://10.100.140.35:7077")
  .getOrCreate()