Delta lake with Java--将数据保存到Minio-编程知识

Delta lake with Java--将数据保存到Minio

今天看了之前发的文章，居然有1条评论，看到我写的东西还是有点用。

今天要解决的问题是如何将 Delta产生的数据保存到Minio里面。

1、安装Minio，去官网下载最新版本的Minio，进入下载目录，运行如下命令，曾经尝试过用docker来安装，不过数据无法保存成功。

minio.exe server D:\bigdata\miniodata --console-address ":9001"，运行结果如下图：

2、登录Minio，建立用来存放数据的桶，记得要设为public访问

3、修改pom.xml增加hadoop-aws依赖，这里要注意版本号，不知道如何确定版本号，去spark下载目录里面的jar目录，找hadoop-client-api-***.jar，其中***就是版本号了

<?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>detla-lake-with-java</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.13</artifactId><version>3.5.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.13</artifactId><version>3.5.1</version></dependency><dependency><groupId>io.delta</groupId><artifactId>delta-spark_2.13</artifactId><version>3.1.0</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.15.2</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-aws</artifactId><version>3.3.4</version></dependency></dependencies></project>

4、新建一个类，命名为DeltaLakeOnMinio，具体代码如下，注意：

config("spark.hadoop.fs.s3a.fast.upload.buffer", "bytebuffer")这一行一定要添加，否则会报错

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;public class DeltaLakeOnMinio {public static void main(String[] args) {SparkSession spark = SparkSession.builder().master("local[*]").appName("delta_lake")
//                .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.S3SingleDriverLogStore")
//                .config("spark.hadoop.fs.s3a.aws.credentials.provider", "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider").config("spark.hadoop.fs.s3a.access.key", "admin").config("spark.hadoop.fs.s3a.secret.key", "password").config("spark.hadoop.fs.s3a.endpoint", "http://127.0.0.1:9000")
//                .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false").config("spark.hadoop.fs.s3a.fast.upload.buffer", "bytebuffer") //一定要加这句否则报错.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension").config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog").getOrCreate();Dataset<Row> data = spark.range(0,5).toDF();data.write().format("delta").mode(SaveMode.Overwrite).save("s3a://delta1/delta-table-data");Dataset<Row> df = spark.read().format("delta").load("s3a://delta1/delta-table-data");df.printSchema();df.show();}}

5、在IDEA运行结果如下图：