Spark Sharing: A Framework for Sharing Data and Resources Across Clusters

 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

<?xml version="1.0" encoding="UTF-8"?>
<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/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>spark-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <scala.version>2.13</scala.version>
        <spark.version>3.3.1</spark.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-graphx_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.22</version>
        </dependency>
    </dependencies>

</project>

 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

import org.apache.spark.sql.*;
import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.apache.spark.sql.types.DataTypes;
import java.util.Properties;
import static org.apache.spark.sql.functions.*;

public class spark01_helloworld {
    public static void main(String[] args){
        SparkSession sparkSession = new SparkSession.Builder()
                .appName("csv2db")
                .master("local[*]")
                .getOrCreate();

        Dataset<Row> bankTable = sparkSession.read()
                .format("csv")
                .option("header", "true")
                .load("C:\\Users\\JiaRui\\Desktop\\bank-additional-full.csv");

        //TODO: 1.过滤掉通话时间少于50秒的数据
        bankTable = bankTable.filter("duration >= 50");

        //TODO: 2.过滤掉年龄小于20
        bankTable.createOrReplaceTempView("bank_table");
        bankTable = sparkSession.sql("select * from bank_table where age >= 20");

        //TODO: 3.生成完整名字，使用逗号连接姓和名字段得出
        bankTable= bankTable.withColumn(
                "name",
                concat(bankTable.col("lname"), lit(", "), bankTable.col("fname"))
        );

        //TODO: 4.生成详细地址字段，使用逗号连接州和城市字段得出
        UserDefinedFunction mergeUdf = udf(
                (String x, String y) -> x + ", " + y, DataTypes.StringType
        );
        sparkSession.udf().register("merge", mergeUdf);
        bankTable = bankTable.withColumn("address",
                call_udf("merge",col("province"),col("city")));

        //TODO: 5.根据年龄将客户分组，并统计各组客户的客户总数、通话联系的平均次数、电话联系的平均时间
        Dataset<Row> reportTable = bankTable.groupBy("age")
                .agg(count("age").name("total customers"),
                        round(avg("campaign"), 2).name("avg_calls"),
                        round(avg("duration"), 2).name("avg_dur"))
                .orderBy("age");

        String url = "jdbc:mysql://172.20.121.222:3306/test";

        Properties properties = new Properties();
        properties.setProperty("dirver","com.mysql.cj.jdbc.Driver");
        properties.setProperty("user","root");
        properties.setProperty("password","123456");

        bankTable.write()
                .mode(SaveMode.Overwrite)
                .jdbc(url,"bank_table",properties);

        reportTable.write()
                .mode(SaveMode.Overwrite)
                .jdbc(url,"report_table",properties);

        sparkSession.stop();
    }
}

 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

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.Trigger;
import static org.apache.spark.sql.functions.*;

public class spark02_streaming {
    public static void main(String[] args) throws Exception {
        SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("StockCCI")
                .getOrCreate();

        //分别设置window长度、容忍最大晚到时间和触发间隔
        String windowDuration = "30 minutes";
        String waterThreshold = "5 minutes";
        String triggerTime = "1 minutes";

        sparkSession.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers","broker1:port1,broker2:port2")
                .option("subcribe","stock")
                .load()
                .selectExpr(
                        "CAST(key AS STRING)",
                        "CAST(value AS STRING)".split(",")[0],
                        "CAST(value AS STRING)".split(",")[1],
                        "CAST(value AS STRING)".split(",")[2],
                        "CAST(value AS STRING)".split(",")[3],
                        "CAST(value AS STRING)".split(",")[4])
                .toDF("companyno","timestamp","price","bidprice","sellpirce","avgprice")
                .selectExpr(
                        "CAST(companyno AS STRING)",
                        "CAST(timestamp AS TIMESTAMP)",
                        "CAST(price AS DOUBLE)",
                        "CAST(bidprice AS DOUBLE)",
                        "CAST(sellpirce AS DOUBLE)",
                        "CAST(avgprice AS DOUBLE)"
                )
                //设定水位
                .withWatermark("timestamp",waterThreshold)
                .groupBy("timestamp",
                        windowDuration,
                        "companyno")
                //求出最高价和最低价
                .agg(
                        max(col("price")).as("max_price"),
                        min(col("price")).as("min_price")
                )
                .writeStream()
                .outputMode("append")
                .trigger(Trigger.ProcessingTime(triggerTime))
                .format("HBaseWriter")
                .start()
                .awaitTermination();
    }
}

 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

import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.feature.HashingTF;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

public class spark03_mlpipeline {
    public static void main(String[] args) {
        SparkSession sparkSession = SparkSession
                .builder()
                .master("local[*]")
                .appName("mlpipeline")
                .getOrCreate();

        Dataset<Row> training = sparkSession.read()
                .format("csv")
                .option("header", "true")
                .schema(new StructType(new StructField[] {
                        new StructField("id", DataTypes.LongType, false, Metadata.empty()),
                        new StructField("text", DataTypes.StringType, false, Metadata.empty()),
                        new StructField("label", DataTypes.DoubleType, false, Metadata.empty())
                }))
                .load("C:\\Users\\JiaRui\\Desktop\\training.csv");

        Tokenizer tokenizer = new Tokenizer()
                .setInputCol("text")
                .setOutputCol("words");

        HashingTF hashingTF = new HashingTF()
                .setNumFeatures(1000)
                .setInputCol(tokenizer.getOutputCol())
                .setOutputCol("features");

        LogisticRegression lr = new LogisticRegression()
                .setMaxIter(10)
                .setRegParam(0.001);

        Pipeline pipeline = new Pipeline()
                .setStages(new PipelineStage[]{tokenizer,hashingTF,lr});

        PipelineModel model = pipeline.fit(training);

        Dataset<Row> test = sparkSession.read()
                .format("csv")
                .option("header", "true")
                .schema(new StructType(new StructField[]{
                        new StructField("id", DataTypes.LongType, false, Metadata.empty()),
                        new StructField("text", DataTypes.StringType, false, Metadata.empty())
                }))
                .load("C:\\Users\\JiaRui\\Desktop\\test.csv");

        Dataset<Row> res = model.transform(test)
                .select("id", "text", "probability", "prediction");

        res.show();
    }
}

 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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.graphx.Graph;
import org.apache.spark.graphx.lib.PageRank;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;
import scala.reflect.ClassTag;

public class spark04_grahpx {
    public static void main(String[] args) {
        SparkSession sparkSession = new SparkSession
                .Builder()
                .appName("pageRank")
                .master("local[*]")
                .getOrCreate();
        sparkSession.sparkContext().setLogLevel("WARN");

        //     URL         neighbor URL
        //     URL         neighbor URL
        //     ...
        JavaRDD<String> lines = sparkSession.read()
                .textFile("C:\\Users\\JiaRui\\Desktop\\pageRank.csv")
                .javaRDD();

        JavaRDD<Tuple2<Object, Object>> edges = lines.map(s -> {
            String[] parts = s.split("\\s+");
            return new Tuple2<>(parts[0], parts[1]);
        });

        Graph<Object, Object> graph = Graph.fromEdgeTuples(edges.rdd(),
                1,
                null,
                StorageLevel.MEMORY_AND_DISK(),
                StorageLevel.MEMORY_AND_DISK(),
                ClassTag.Any());

        Graph<Object, Object> rankedGraph = PageRank.runUntilConvergence(graph, 0.001, 0.001, ClassTag.Any(), ClassTag.Any());

        JavaRDD<Tuple2<Object, Object>> pageRankValues = rankedGraph.vertices().toJavaRDD();

        pageRankValues.foreach(x -> System.out.println(x));
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

//读取 json 文件
spark.read.json("examples/src/main/resources/people.json");
//读取 csv 文件
spark.read.csv("examples/src/main/resources/people.csv");
//读取 parquet 文件
spark.read.parquet("examples/src/main/resources/people.parquet");
//读取 orc 文件
spark.read.orc("examples/src/main/resources/people.orc");
//读取文本文件
spark.read.text("examples/src/main/resources/people.csv");
//通过 JDBC 连接外部数据库生成
spark.read.format("jdbc")
          .option("url","jdbc:driver")
          .option("user","")
          .option("password","")
          .load();

1
2
3
4
5
6
7
8
9

//条件查询
df.select("age").where("name is not null and age > 10").foreach(println(_));

//分组统计
df.groupBy("age").count().foreach(println(_));

//连接操作: 
//支持 inner/cross/outer/full/full_outer/left/left_outer/cross/left_semi/left_anti等
leftDF.join(rightDF, leftDF("pid") == rightDF("fid"),"left_outer");

1
2
3
4
5
6
7

SparkSession sparkSession = new SparkSession
                .Builder()
                .appName("newApp")
                .master("yarn-client")
                .config("spark.executor.instances", "10")
                .config("spark.executor.memory", "10g")
                .getOrCreate();