Spark 读 S3 Parquet 写入 Hudi 表

目录

Spark 读 S3 Parquet 写入 Hudi 表

参考

关于S3，S3N和S3A的区别与联系

Spark 读写 S3 Parquet 文件

测试代码

pom.xml

配置文件

EMR Spark任务提交

spark-shell

spark-submit

Spark 读写 Hudi

本地测试

代码

集群上测试

spark-shell

spark-sql

Spark-submit

Hive 中测试

问题解决

参考

Hadoop-aws

Apache Hadoop Amazon Web Services support – Hadoop-AWS module: Integration with Amazon Web Services

EMR 对应版本

Amazon EMR release 6.5.0 - Amazon EMR

EMR Spark

Apache Spark - Amazon EMR

EMR Hudi

Hudi - Amazon EMR

关于S3，S3N和S3A的区别与联系

首先是三种协议的访问大小有区别；

其次S3是block-based，s3n/s3a是 object-based。

最后S3A是apache推荐的访问方式，且S3访问方式将会慢慢被替代，AWS不赞成使用S3访问，且S3A更加稳定安全高效

S3 Block FileSystem (URI scheme: s3) A block-based filesystem backed by S3. Files are stored as blocks, just like they are in HDFS. This permits efficient implementation of renames. This filesystem requires you to dedicate a bucket for the filesystem - you should not use an existing bucket containing files, or write other files to the same bucket. The files stored by this filesystem can be larger than 5GB, but they are not interoperable with other S3 tools.

S3A (URI scheme: s3a) A successor to the S3 Native, s3n fs, the S3a: system uses Amazon's libraries to interact with S3. This allows S3a to support larger files (no more 5GB limit), higher performance operations and more. The filesystem is intended to be a replacement for /successor to S3 Native: all objects accessible from s3n:// URLs should also be accessible from s3a simply by replacing the URL schema.

S3 Native FileSystem (URI scheme: s3n) A native filesystem for reading and writing regular files on S3. The advantage of this filesystem is that you can access files on S3 that were written with other tools. Conversely, other tools can access files written using Hadoop. The disadvantage is the 5GB limit on file size imposed by S3.

Spark 读写 S3 Parquet 文件

测试代码

package org.zero
​
import com.amazonaws.auth.{ClasspathPropertiesFileCredentialsProvider, DefaultAWSCredentialsProviderChain}
import org.apache.log4j.{Level, Logger}
import org.slf4j.LoggerFactory
import org.utils.SparkUtils
import software.amazon.awssdk.auth.credentials.{EnvironmentVariableCredentialsProvider, ProfileCredentialsProvider}
​
object SparkS2Test {private var logger: org.slf4j.Logger = _
​def main(args: Array[String]): Unit = {logger = LoggerFactory.getLogger(this.getClass.getSimpleName)Logger.getLogger("org.apache.hadoop").setLevel(Level.INFO)Logger.getLogger("org.apache.spark").setLevel(Level.INFO)Logger.getLogger("org.spark_project.jetty").setLevel(Level.WARN)
​val start = System.currentTimeMillis()logger.warn(s"=================== Spark 读取 S3 ===================")
​val spark = SparkUtils.getSparkSession(this.getClass.getSimpleName, "local[*]")val sc = spark.sparkContextsc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")
​val dataframe = spark.read.parquet("s3a://s3-datafacts-poc-001/dct/s3-datafacts-poc-001/dt=2022-05-09")
​val tmpCache = dataframe.cache()tmpCache.createOrReplaceTempView("parquet_tmp_view")
​val dataFrame2 = spark.sql("select * from parquet_tmp_view limit 10")
​dataFrame2.show
​
//    dataFrame2.write.parquet("F:\\tmp\\output")
​spark.stop()
​val end = System.currentTimeMillis()logger.warn(s"=================== 耗时： ${(end - start) / 1000} 秒 ===================")}
}
package org.utils
​
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, SparkSession}
​
object SparkUtils {private val sparkConf: SparkConf = new SparkConf()
​def getSparkConf(appName: String, master: String): SparkConf = {sparkConf.setMaster(master).setAppName(appName)}
​def getSparkSession(appName: String, master: String): SparkSession = {sparkConf.setMaster(master).setAppName(appName)sparkSessionInit}
​lazy val sparkSessionInit: SparkSession = SparkSession.builder().config(sparkConf).config("spark.serializer", "org.apache.spark.serializer.KryoSerializer").config("spark.io.compression.codec", "snappy").config("spark.rdd.compress", "true").config("spark.hadoop.parquet.writer.version", "v2").config("spark.sql.parquet.enableVectorizedReader", "false").config("spark.sql.parquet.compression.codec", "false").config("spark.sql.parquet.compression.codec", "snappy").config("spark.sql.parquet.filterPushdown", "true").config("spark.sql.parquet.mergeSchema", "true").config("spark.sql.parquet.binaryAsString", "true").getOrCreate()
}

pom.xml

<?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-s3-hudi-test</artifactId><version>1.0-SNAPSHOT</version>
​<name>spark-s3-hudi-test</name>
​<properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><scala.maven.plugin.version>4.3.0</scala.maven.plugin.version><maven.compiler.plugin.version>3.8.1</maven.compiler.plugin.version><maven.assembly.plugin.version>3.1.1</maven.assembly.plugin.version><scala.version>2.12.13</scala.version><scala.binary.version>2.12</scala.binary.version><spark.version>3.1.2</spark.version><hadoop.version>3.2.1</hadoop.version><fasterxml.jackson.version>2.10.0</fasterxml.jackson.version><project.build.scope>compile</project.build.scope></properties>
​<repositories><repository><id>emr-6.5.0-artifacts</id><name>EMR 6.5.0 Releases Repository</name><releases><enabled>true</enabled></releases><snapshots><enabled>false</enabled></snapshots><url>https://s3.us-west-1.amazonaws.com/us-west-1-emr-artifacts/emr-6.5.0/repos/maven/</url></repository></repositories><dependencyManagement><dependencies><dependency><groupId>software.amazon.awssdk</groupId><artifactId>bom</artifactId><version>2.17.186</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement>
​<dependencies><dependency><groupId>software.amazon.awssdk</groupId><artifactId>s3</artifactId></dependency><dependency><groupId>software.amazon.awssdk</groupId><artifactId>kms</artifactId></dependency><dependency><groupId>software.amazon.awssdk</groupId><artifactId>s3control</artifactId></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${spark.pom.scope}</scope></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-aws</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>${fasterxml.jackson.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>${fasterxml.jackson.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-annotations</artifactId><version>${fasterxml.jackson.version}</version></dependency><dependency><groupId>org.apache.parquet</groupId><artifactId>parquet-avro</artifactId><version>1.12.0</version></dependency><dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpcore</artifactId><version>4.4.15</version></dependency><dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient</artifactId><version>4.5.13</version></dependency></dependencies>
​<build><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>${scala.maven.plugin.version}</version><executions><execution><goals><goal>compile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>${maven.assembly.plugin.version}</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>
</project>

配置文件

创建 resources 目录，添加配置文件

core-site.xml

<configuration><property><name>fs.s3a.aws.credentials.provider</name><value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value></property>
<!--    <property>-->
<!--        <name>fs.s3a.access.key</name>-->
<!--        <description>AWS access key ID.-->
<!--            Omit for IAM role-based or provider-based authentication.</description>-->
<!--        <value>AKIA4ZNT6QH3L45V45VY</value>-->
<!--    </property>-->
<!--    <property>-->
<!--        <name>fs.s3a.secret.key</name>-->
<!--        <description>AWS secret key.-->
<!--            Omit for IAM role-based or provider-based authentication.</description>-->
<!--        <value>og8I6vB52vDhhb/So/r9ioHMvtbJ4EI2xdGPQIce</value>-->
<!--    </property>-->
</configuration>

log4j.properties

################################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
log4j.rootLogger=info, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
#log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p - %m%n
log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-10c %x - %m%n

EMR Spark任务提交

spark-shell

spark-shell \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.hive.convertMetastoreParquet=false" \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "false")
spark.conf.set("spark.sql.parquet.binaryAsString", "true")
​
sc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")
sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")
sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")val df = spark.read.parquet("s3a://s3-datafacts-poc-001/dw/ods/test_s3_01/dt=2022-05-11")
df.show

spark-submit

将代码打包上传，不需要打包依赖

spark-submit \
--master local[2] \
--class org.zero.SparkS2Test \
/opt/jars/spark-s3-hudi-test-1.0-SNAPSHOT.jar
spark-submit \
--master yarn \
--deploy-mode client \
--executor-memory 1G \
--total-executor-cores 2 \
--class org.zero.SparkS2Test \
/opt/jars/spark-s3-hudi-test-1.0-SNAPSHOT.jar
spark-submit \
--master yarn \
--deploy-mode client \
--num-executors 2 \
--executor-cores 2 \
--executor-memory 1g \
--class org.zero.SparkS2Test \
s3://s3-datafacts-poc-001/spark-s3-hudi-test-1.0-SNAPSHOT.jar

Spark 读写 Hudi

本地测试

如果使用AWS hudi.jar，从msater服务器下载 hudi 包，并导入本地 maven 仓库

/usr/lib/hudi/hudi-spark3-bundle_2.12-0.9.0-amzn-1.jar
​
mvn install:install-file -Dfile=D:\soft\hudi-spark3-bundle_2.12-0.9.0-amzn-1.jar -DgroupId=org.apache.hudi -DartifactId=hudi-spark3-bundle_2.12 -Dversion=0.9.0-amzn-1 -Dpackaging=jar

项目中添加依赖

        <dependency><groupId>com.amazon.awssdk</groupId><artifactId>hudi-spark3-bundle_${scala.binary.version}</artifactId><version>0.9.0-amzn-1</version></dependency>

代码

package org.zero
​
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieIndexConfig._
import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
import org.apache.hudi.hive.MultiPartKeysValueExtractor
import org.apache.hudi.index.HoodieIndex.IndexType
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions.{current_date, lit}
import org.slf4j.LoggerFactory
import org.utils.SparkUtils
​
​
object SparkS3HudiTest {private var logger: org.slf4j.Logger = _
​def main(args: Array[String]): Unit = {logger = LoggerFactory.getLogger(this.getClass.getSimpleName)Logger.getLogger("org.apache.hadoop").setLevel(Level.INFO)Logger.getLogger("org.apache.spark").setLevel(Level.INFO)Logger.getLogger("org.spark_project.jetty").setLevel(Level.DEBUG)
​val start = System.currentTimeMillis()
​
​val spark = SparkUtils.getSparkSession(this.getClass.getSimpleName, "local[1]")val sc = spark.sparkContextsc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")//    sc.hadoopConfiguration.set("fs.s3.connection.ssl.enabled", "false")
​// 读 s3//    val dataframe = spark//      .read//      .parquet("s3a://s3-datafacts-poc-001/dct/s3-datafacts-poc-001/dt=2022-05-09")
​//    val tmpCache = dataframe.cache()//    tmpCache.createOrReplaceTempView("parquet_tmp_view")////    val dataFrame2 = spark.sql("select * from parquet_tmp_view limit 10")////    dataFrame2.show
​// 写 s3//        val dataframe = spark//          .read//          .parquet("F:\\tmp\\test_table.parquet")////        dataframe//          .write//          .mode(SaveMode.Overwrite)//          .save("s3a://s3-datafacts-poc-001/test/test_parquet")
​// 读 s3 多对象//    logger.warn(s"=================== Spark 读 S3 Parquet ===================")//    val dataframe2 = spark.read//      .parquet("s3a://s3-datafacts-poc-001/dw/ods/test_s3_01/dt=2022-05-1[1-3]")//    dataframe2.printSchema()//    dataframe2.show()//    val dataframe3 = dataframe2.withColumn("dt", current_date())//    dataframe2.withColumn("last_update_time" , unix_timestamp())
​// Spark Dataframe 写 Hudi//    logger.warn(s"=================== Hudi 参数配置 ===================")// hudi 参数设置//    val hudiOptions = Map[String, String](//      HoodieWriteConfig.TBL_NAME.key() -> "hudi_test_table",//      TABLE_TYPE.key() -> COW_TABLE_TYPE_OPT_VAL,//      RECORDKEY_FIELD.key() -> "id",//      PARTITIONPATH_FIELD.key() -> "dt",//      PRECOMBINE_FIELD.key() -> "time",//      BLOOM_INDEX_UPDATE_PARTITION_PATH_ENABLE.key() -> "true",//      INDEX_TYPE.key()-> IndexType.BLOOM.name(),//      META_SYNC_ENABLED.key() -> "true",//      HIVE_SYNC_ENABLED.key() -> "true",//      HIVE_USER.key() -> "hive",//      HIVE_PASS.key() -> "hive",//      HIVE_DATABASE.key() -> "ods",//      HIVE_TABLE.key() -> "hudi_test_table",//      HIVE_URL.key() -> "jdbc:hive2://52.82.123.13:10000",//      //      HIVE_URL.key() -> "jdbc:hive2://172.31.194.132:10000",//      HIVE_PARTITION_FIELDS.key() -> "dt",//      HIVE_PARTITION_EXTRACTOR_CLASS.key() -> classOf[MultiPartKeysValueExtractor].getName,//      HIVE_AUTO_CREATE_DATABASE.key() -> "true"//    )//    println("===> \n" + hudiOptions)
​//    logger.warn(s"=================== Spark 写 S3 Hudi ===================")//    dataframe3.write//      .format("org.apache.hudi")//      .option(OPERATION.key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)//      .options(hudiOptions)//      .mode(SaveMode.Overwrite)//      .save("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table")
​// spark 读 hudi//    logger.warn(s"=================== Spark 读取 S3 Hudi ===================")//    val readDF = spark.read//      .format("org.apache.hudi")//      .load("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table")//      .select("*")//      .sort("creation_date","id")//      .show(10)
​// Spark SQL 读 Hive 写 Hudilogger.warn(s"=================== SparkSQL 创建 Hive 表 ===================")spark.sql("set spark.hadoop.hive.exec.dynamic.partition.mode=nonstrict")spark.sql("set spark.hadoop.hive.exec.compress.output=true")spark.sql("set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec")spark.sql("set mapreduce.output.fileoutputformat.compress.type=BLOCK")
​//    val hiveSourceDDL =//      """//        |create external table if not exists ods.test_parquet_to_hive_table(//        | id int,//        | name string,//        | age int,//        | job string,//        | address string,//        | company string,//        | email string,//        | url string,//        | phone string,//        | sfzh string,//        | chrome string,//        | ipv4 string,//        | ipv6 string,//        | `date` bigint,//        | `time` bigint,//        | mac_address string,//        | col_tinyint int,//        | col_smallint int,//        | col_mediumint int,//        | col_bigint bigint,//        | col_decimal double,//        | col_double double,//        | col_float double,//        | col_time bigint,//        | col_blob string,//        | col_text string//        |) partitioned by(dt string)//        |stored as parquet//        |location "s3://s3-datafacts-poc-001/dw/ods/test_parquet_to_hive_table"//        |""".stripMargin//    spark.sql(hiveSourceDDL)
​//    spark.sql("show databases").showspark.sql("use ods")//    spark.sql("show tables").show//    spark.sql("show create table test_parquet_to_hive_table").show
​logger.warn(s"=================== SparkSQL 创建 Hudi Sink 表 ===================")
​val hudiSinkDDL ="""|create table if not exists ods.ods_hudi_sink_table (| id int,| name string,| age int,| job string,| address string,| company string,| email string,| url string,| phone string,| sfzh string,| chrome string,| ipv4 string,| ipv6 string,| `date` bigint,| `time` bigint,| mac_address string,| col_tinyint int,| col_smallint int,| col_mediumint int,| col_bigint bigint,| col_decimal double,| col_double double,| col_float double,| col_time bigint,| col_blob string,| col_text string,| dt date|) using hudi|partitioned by (dt)|tblproperties (|  type = 'cow',|  primaryKey = 'id',|  preCombineField = 'time',|  hoodie.index.type = 'GLOBAL_BLOOM',|  hiveSyncEnabled = 'true',|  hiveDatabase = 'ods',|  hiveUser = 'hive',|  hivePass = 'hive',|  hiveTable = 'ods_hudi_sink_table',|  hiveUrl = 'jdbc:hive2://52.82.123.13:10000',|  hivePartitionFields = 'dt'|)|location "s3a://s3-datafacts-poc-001/dw/ods/ods_hudi_sink_table";|""".stripMargin
​spark.sql(hudiSinkDDL)spark.sql("show tables").show
​logger.warn(s"=================== SparkSQL 读 Hive 写入 Hudi 表 ===================")spark.sql("insert into table ods_hudi_sink_table select * from test_parquet_to_hive_table")
​spark.sql("select * from ods_hudi_sink_table limit 10").show
​spark.stop()
​val end = System.currentTimeMillis()logger.warn(s"=================== 耗时： ${(end - start) / 1000} 秒 ===================")}
}

pom.xml

<?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-s3-hudi-test</artifactId><version>1.0-SNAPSHOT</version><name>spark-s3-hudi-test</name><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><scala.maven.plugin.version>4.3.0</scala.maven.plugin.version><maven.compiler.plugin.version>3.8.1</maven.compiler.plugin.version><maven.assembly.plugin.version>3.1.1</maven.assembly.plugin.version><scala.version>2.12.13</scala.version><scala.binary.version>2.12</scala.binary.version><spark.version>3.1.2</spark.version><hadoop.version>3.2.1</hadoop.version><hoodie.version>0.9.0</hoodie.version>
<!--        <hoodie.version>0.9.0-amzn-1</hoodie.version>--><fasterxml.jackson.version>2.10.5</fasterxml.jackson.version><project.build.scope>compile</project.build.scope>
<!--        <project.build.scope>provided</project.build.scope>--></properties><dependencies><!-- s3 --><dependency><groupId>com.amazonaws</groupId><artifactId>aws-java-sdk-s3</artifactId><version>1.12.217</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>com.amazonaws</groupId><artifactId>aws-java-sdk-s3control</artifactId><version>1.12.217</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>com.amazonaws</groupId><artifactId>aws-java-sdk-kms</artifactId><version>1.12.217</version><scope>${project.build.scope}</scope></dependency><!-- scala --><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version><scope>${project.build.scope}</scope></dependency><!-- spark --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-avro_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency><!-- hadoop --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-aws</artifactId><version>${hadoop.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version><scope>${project.build.scope}</scope></dependency><!-- hudi --><dependency><groupId>org.apache.hudi</groupId><artifactId>hudi-spark3-bundle_${scala.binary.version}</artifactId><version>${hoodie.version}</version><scope>${project.build.scope}</scope></dependency><!-- jackson --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>${fasterxml.jackson.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>${fasterxml.jackson.version}</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-annotations</artifactId><version>${fasterxml.jackson.version}</version><scope>${project.build.scope}</scope></dependency><!-- http --><dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpcore</artifactId><version>4.4.11</version><scope>${project.build.scope}</scope></dependency><dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient</artifactId><version>4.5.9</version><scope>${project.build.scope}</scope></dependency></dependencies><build><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>${scala.maven.plugin.version}</version><executions><execution><goals><goal>compile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>${maven.assembly.plugin.version}</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>
</project>

集群上测试

spark-shell

spark-shell \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.hive.convertMetastoreParquet=false" \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar

配置

spark.conf.set("spark.sql.parquet.enableVectorizedReader", "false")
spark.conf.set("spark.sql.parquet.binaryAsString", "true")
​
sc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")
sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")
sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")
sc.hadoopConfiguration.set("fs.s3a.fast.upload", "true")

导入包

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hive.MultiPartKeysValueExtractor

读 S3 parquet 文件写入 S3 Hudi 表

val df = spark.read.parquet("s3a://s3-datafacts-poc-001/dw/ods/test_s3_01/dt=2022-05-1[1-3]")
​
val df2 = df.withColumn("creation_date" , current_date())
​
val hudiOptions = Map[String, String](HoodieWriteConfig.TBL_NAME.key()  -> "hudi_test_table",TABLE_TYPE.key() -> COW_TABLE_TYPE_OPT_VAL,RECORDKEY_FIELD.key() -> "id",                    // 主键PARTITIONPATH_FIELD.key() -> "creation_date",     // 分区PRECOMBINE_FIELD.key() -> "time",                 // 数据更新时间的列名BLOOM_INDEX_UPDATE_PARTITION_PATH_ENABLE.key() -> "true",     //当分区变更时，当前数据的分区目录是否变更INDEX_TYPE.key()-> IndexType.BLOOM.name(),        // 设置索引类型目前有HBASE,INMEMORY,BLOOM,GLOBAL_BLOOM四种索引为了保证分区变更后能找到必须设置全局GLOBAL_BLOOMMETA_SYNC_ENABLED.key() -> "true",                // 同步元数据HIVE_SYNC_ENABLED.key() -> "true",                // 同步数据到HiveHIVE_USER.key() -> "hive",HIVE_PASS.key() -> "hive",HIVE_DATABASE.key() -> "ods",HIVE_TABLE.key()-> "hudi_test_table",HIVE_URL.key() -> "jdbc:hive2://172.31.194.132:10000",HIVE_PARTITION_FIELDS.key() -> "creation_date",HIVE_PARTITION_EXTRACTOR_CLASS.key() ->  classOf[MultiPartKeysValueExtractor].getName,HIVE_AUTO_CREATE_DATABASE.key() -> "true"
)
​
(df2.write.format("org.apache.hudi").option(OPERATION.key(), DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL).options(hudiOptions).mode(SaveMode.Overwrite).save("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table"))
​
val df3 = spark.read.format("org.apache.hudi").load("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table").select("*").sort("creation_date", "id")df3.show(10)

更新

val updateDF = df2.limit(1).withColumn("creation_date", lit("date"))
​
(updateDF.write.format("org.apache.hudi").option(OPERATION.key(), DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL).options(hudiOptions).mode(SaveMode.Append).save("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table"))updateDF.select()

s3文件删除


aws s3 rm se3://s3-datafacts-poc-001/dw/ods/hudi_test_tableemrfs sync se3://s3-datafacts-poc-001/dw/ods

Spark-shell 中使用 SparkSql

spark-shell \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.hive.convertMetastoreParquet=false" \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/hive/lib/mysql-connector-java-5.1.49.jar

配置

spark.conf.set("spark.sql.parquet.enableVectorizedReader", "false")
spark.conf.set("spark.sql.parquet.binaryAsString", "true")
​
sc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")
sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")
sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")
sc.hadoopConfiguration.set("fs.s3a.fast.upload", "true")

S3 Parquet文件导入Hive

spark.sql("show databases").show
​
+---------+
|namespace|
+---------+
|  default|
| dhw_demo|
|      ods|
|    test1|
|  test_db|
|     yszy|
+---------+

Parquet 数据源

val df = spark.read.parquet("s3a://s3-datafacts-poc-001/dw/ods/test_s3_01/dt=2022-05-1[1-3]")
​
val df2 = df.withColumn("dt" , current_date())
​
df2.printSchema
​
root|-- id: integer (nullable = true)|-- name: string (nullable = true)|-- age: integer (nullable = true)|-- job: string (nullable = true)|-- address: string (nullable = true)|-- company: string (nullable = true)|-- email: string (nullable = true)|-- url: string (nullable = true)|-- phone: string (nullable = true)|-- sfzh: string (nullable = true)|-- chrome: string (nullable = true)|-- ipv4: string (nullable = true)|-- ipv6: string (nullable = true)|-- date: long (nullable = true)|-- time: long (nullable = true)|-- mac_address: string (nullable = true)|-- col_tinyint: integer (nullable = true)|-- col_smallint: integer (nullable = true)|-- col_mediumint: integer (nullable = true)|-- col_bigint: long (nullable = true)|-- col_decimal: double (nullable = true)|-- col_double: double (nullable = true)|-- col_float: double (nullable = true)|-- col_time: long (nullable = true)|-- col_blob: string (nullable = true)|-- col_text: string (nullable = true)|-- dt: date (nullable = false)spark.sql("set spark.sql.debug.maxToStringFields=100")
df2.createOrReplaceTempView("parquet_tmp_table")
​
spark.sql("use ods")
spark.sql("show tables").show
​
spark.sql("desc formatted parquet_tmp_table").showspark.sql("select * from parquet_tmp_table limit 10").show

Hive 建表 Sink表，当然，也可以在 spark-sql 中创建

beeline -n hive -u jdbc:hive2://52.82.123.13:10000/default
​
SET hive.exec.compress.output=true;
SET mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
SET mapred.output.compression.type=BLOCK;
​
# alter table test_parquet_to_hive_table set tblproperties('external'='true');
drop table if exists test_parquet_to_hive_table;
​
create external table if not exists ods.test_parquet_to_hive_table(id int,name string,age int,job string,address string,company string,email string,url string,phone string,sfzh string,chrome string,ipv4 string,ipv6 string,`date` bigint,`time` bigint,mac_address string,col_tinyint int,col_smallint int,col_mediumint int,col_bigint bigint,col_decimal double,col_double double,col_float double,col_time bigint,col_blob string,col_text string
) partitioned by(dt string)
stored as parquet
location "s3://s3-datafacts-poc-001/dw/ods/test_parquet_to_hive_table";

写入Hive表

spark.sql("show create table ods.test_parquet_to_hive_table").show
​
spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
spark.sql("set spark.hadoop.hive.exec.dynamic.partition.mode=nonstrict")
​
spark.sql("insert overwrite ods.test_parquet_to_hive_table select * from parquet_tmp_table")
spark.sql("select * from ods.test_parquet_to_hive_table limit 10").show

读取 Hive 表写入Hudi

spark-sql

spark-sql \
--conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
--conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/hive/lib/mysql-connector-java-5.1.49.jar
set spark.sql.debug.maxToStringFields=100;
set fs.s3a.access.key=xxxx;
set fs.s3a.secret.key=xxxx;
set fs.s3a.endpoint=s3.cn-northwest-1.amazonaws.com.cn;

创建 Hudi 表

create table if not exists ods.ods_hudi_sink_table (id int,name string,age int,job string,address string,company string,email string,url string,phone string,sfzh string,chrome string,ipv4 string,ipv6 string,`date` bigint,`time` bigint,mac_address string,col_tinyint int,col_smallint int,col_mediumint int,col_bigint bigint,col_decimal double,col_double double,col_float double,col_time bigint,col_blob string,col_text string,dt date
) using hudi
partitioned by (dt)
tblproperties (type = 'cow',primaryKey = 'id',preCombineField = 'time',hoodie.index.type = 'GLOBAL_BLOOM',hiveSyncEnabled = 'true',hiveDatabase = 'ods',hiveUser = 'hive',hivePass = 'hive',hiveTable = 'ods_hudi_sink_table',hiveUrl = 'jdbc:hive2://52.82.123.13:10000',hivePartitionFields = 'dt'
)
location "s3://s3-datafacts-poc-001/dw/ods/ods_hudi_sink_table";

写入 Hudi

show tables;
​
#ods     hudi_test_table false
#ods     ods_hudi_sink_table     false
#ods     test_parquet_to_hive_table      false
#ods     test_tb_02      false
#Time taken: 0.021 seconds, Fetched 4 row(s)
​
insert into table ods_hudi_sink_table select * from test_parquet_to_hive_table;select * from ods_hudi_sink_table limit 10;


Spark-submit


spark-submit \
--master local[2] \
--class org.zero.SparkS3HudiTest \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar \
/opt/jars/spark-s3-hudi-test-1.0-SNAPSHOT.jarspark-submit \
--master yarn \
--deploy-mode client \
--executor-memory 1G \
--total-executor-cores 2 \
--class org.zero.SparkS2Test \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar \
/opt/jars/spark-s3-hudi-test-1.0-SNAPSHOT.jar


Hive 中测试


beeline -n hive -u jdbc:hive2://52.82.123.13:10000/default
​
use ods;
​
show tables;
+-----------------------------+
|          tab_name           |
+-----------------------------+
| hudi_test_table             |
| ods_hudi_sink_table         |
| test_parquet_to_hive_table  |
+-----------------------------+
​
show create table hudi_test_table;
+----------------------------------------------------+
|                   createtab_stmt                   |
+----------------------------------------------------+
| CREATE EXTERNAL TABLE `hudi_test_table`(           |
|   `_hoodie_commit_time` string,                    |
|   `_hoodie_commit_seqno` string,                   |
|   `_hoodie_record_key` string,                     |
|   `_hoodie_partition_path` string,                 |
|   `_hoodie_file_name` string,                      |
|   `id` int,                                        |
|   `name` string,                                   |
|   `age` int,                                       |
|   `job` string,                                    |
|   `address` string,                                |
|   `company` string,                                |
|   `email` string,                                  |
|   `url` string,                                    |
|   `phone` string,                                  |
|   `sfzh` string,                                   |
|   `chrome` string,                                 |
|   `ipv4` string,                                   |
|   `ipv6` string,                                   |
|   `date` bigint,                                   |
|   `time` bigint,                                   |
|   `mac_address` string,                            |
|   `col_tinyint` int,                               |
|   `col_smallint` int,                              |
|   `col_mediumint` int,                             |
|   `col_bigint` bigint,                             |
|   `col_decimal` double,                            |
|   `col_double` double,                             |
|   `col_float` double,                              |
|   `col_time` bigint,                               |
|   `col_blob` string,                               |
|   `col_text` string)                               |
| PARTITIONED BY (                                   |
|   `creation_date` date)                            |
| ROW FORMAT SERDE                                   |
|   'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'  |
| WITH SERDEPROPERTIES (                             |
|   'hoodie.query.as.ro.table'='false',              |
|   'path'='s3://s3-datafacts-poc-001/dw/ods/hudi_test_table')  |
| STORED AS INPUTFORMAT                              |
|   'org.apache.hudi.hadoop.HoodieParquetInputFormat'  |
| OUTPUTFORMAT                                       |
|   'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' |
| LOCATION                                           |
|   's3://s3-datafacts-poc-001/dw/ods/hudi_test_table' |
| TBLPROPERTIES (                                    |
|   'bucketing_version'='2',                         |
|   'last_commit_time_sync'='20220513101907',        |
|   'spark.sql.sources.provider'='hudi',             |
|   'spark.sql.sources.schema.numPartCols'='1',      |
|   'spark.sql.sources.schema.numParts'='1',         |
|   'spark.sql.sources.schema.part.0'='{"type":"struct","fields":[{"name":"_hoodie_commit_time","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_commit_seqno","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_record_key","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_partition_path","type":"string","nullable":true,"metadata":{}},{"name":"_hoodie_file_name","type":"string","nullable":true,"metadata":{}},{"name":"id","type":"integer","nullable":true,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"age","type":"integer","nullable":true,"metadata":{}},{"name":"job","type":"string","nullable":true,"metadata":{}},{"name":"address","type":"string","nullable":true,"metadata":{}},{"name":"company","type":"string","nullable":true,"metadata":{}},{"name":"email","type":"string","nullable":true,"metadata":{}},{"name":"url","type":"string","nullable":true,"metadata":{}},{"name":"phone","type":"string","nullable":true,"metadata":{}},{"name":"sfzh","type":"string","nullable":true,"metadata":{}},{"name":"chrome","type":"string","nullable":true,"metadata":{}},{"name":"ipv4","type":"string","nullable":true,"metadata":{}},{"name":"ipv6","type":"string","nullable":true,"metadata":{}},{"name":"date","type":"long","nullable":true,"metadata":{}},{"name":"time","type":"long","nullable":true,"metadata":{}},{"name":"mac_address","type":"string","nullable":true,"metadata":{}},{"name":"col_tinyint","type":"integer","nullable":true,"metadata":{}},{"name":"col_smallint","type":"integer","nullable":true,"metadata":{}},{"name":"col_mediumint","type":"integer","nullable":true,"metadata":{}},{"name":"col_bigint","type":"long","nullable":true,"metadata":{}},{"name":"col_decimal","type":"double","nullable":true,"metadata":{}},{"name":"col_double","type":"double","nullable":true,"metadata":{}},{"name":"col_float","type":"double","nullable":true,"metadata":{}},{"name":"col_time","type":"long","nullable":true,"metadata":{}},{"name":"col_blob","type":"string","nullable":true,"metadata":{}},{"name":"col_text","type":"string","nullable":true,"metadata":{}},{"name":"creation_date","type":"date","nullable":false,"metadata":{}}]}',  |
|   'spark.sql.sources.schema.partCol.0'='creation_date',  |
|   'transient_lastDdlTime'='1652437162')            |
+----------------------------------------------------+select * from hudi_test_table limit 10;


问题解决


问题1


2022-05-10 14:27:07,535 DEBUG org.apache.hadoop.fs.s3a.AWSCredentialProviderList - No credentials provided by SimpleAWSCredentialsProvider: org.apache.hadoop.fs.s3a.CredentialInitializationException: Access key or secret key is unset org.apache.hadoop.fs.s3a.CredentialInitializationException: Access key or secret key is unset at org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider.getCredentials(SimpleAWSCredentialsProvider.java:68) at org.apache.hadoop.fs.s3a.AWSCredentialProviderList.getCredentials(AWSCredentialProviderList.java:137) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.getCredentialsFromContext(AmazonHttpClient.java:1166) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.runBeforeRequestHandlers(AmazonHttpClient.java:762) at ...2022-05-10 14:27:07,535 DEBUG org.apache.hadoop.fs.s3a.AWSCredentialProviderList - No credentials provided by EnvironmentVariableCredentialsProvider: com.amazonaws.SdkClientException: Unable to load AWS credentials from environment variables (AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY)) com.amazonaws.SdkClientException: Unable to load AWS credentials from environment variables (AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY)) at com.amazonaws.auth.EnvironmentVariableCredentialsProvider.getCredentials(EnvironmentVariableCredentialsProvider.java:50) at org.apache.hadoop.fs.s3a.AWSCredentialProviderList.getCredentials(AWSCredentialProviderList.java:137) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.getCredentialsFromContext(AmazonHttpClient.java:1166) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.runBeforeRequestHandlers(AmazonHttpClient.java:762) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:724) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:717) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:699) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:667) at ...


方法1：


配置


    <property><name>fs.s3a.aws.credentials.provider</name><value>com.amazonaws.auth.profile.ProfileCredentialsProvider</value></property>


Linux


~/.aws/credentials
[default]
aws_access_key_id = your_access_key_id
aws_secret_access_key = your_secret_access_key
​
export AWS_ACCESS_KEY_ID=your_access_key_id
export AWS_SECRET_ACCESS_KEY=your_secret_access_key


Windows


C:\Users\LJH\.aws\credentials
[default]
aws_access_key_id = your_access_key_id
aws_secret_access_key = your_secret_access_key
​
set AWS_ACCESS_KEY_ID=AKIA4ZNT6QH3L45V45VY
set AWS_SECRET_ACCESS_KEY=og8I6vB52vDhhb/So/r9ioHMvtbJ4EI2xdGPQIce


方法2：


在 resources 下创建配置文件 core-site.xml


<configuration><property><name>fs.s3a.aws.credentials.provider</name><value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value></property><property><name>fs.s3a.access.key</name><description>AWS access key ID.Omit for IAM role-based or provider-based authentication.</description><value>AKIA4ZNT6QH3L45V45VY</value></property><property><name>fs.s3a.secret.key</name><description>AWS secret key.Omit for IAM role-based or provider-based authentication.</description><value>og8I6vB52vDhhb/So/r9ioHMvtbJ4EI2xdGPQIce</value></property>
</configuration>


方法3：


<configuration><property><name>fs.s3a.aws.credentials.provider</name><value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value></property>
val sc = spark.sparkContext
sc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")
sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")


问题2


Exception in thread "main" org.apache.hadoop.fs.s3a.AWSRedirectException: getFileStatus on s3a://s3-datafacts-poc-001/dct/s3-datafacts-poc-001/dt=2022-05-09: com.amazonaws.services.s3.model.AmazonS3Exception: The bucket is in this region: cn-northwest-1. Please use this region to retry the request (Service: Amazon S3; Status Code: 301; Error Code: 301 Moved Permanently; Request ID: C1AV4REN3P1VGWY7; S3 Extended Request ID: 61G1/2kXI1/mqclSY6UUjFTzm4ANAtuqMiInnat0VZi/8vYVmyxkZ4GzuTnmjeAcwD02yWHl2eE=), S3 Extended Request ID: 61G1/2kXI1/mqclSY6UUjFTzm4ANAtuqMiInnat0VZi/8vYVmyxkZ4GzuTnmjeAcwD02yWHl2eE=:301 Moved Permanently: The bucket is in this region: cn-northwest-1. Please use this region to retry the request (Service: Amazon S3; Status Code: 301; Error Code: 301 Moved Permanently; Request ID: C1AV4REN3P1VGWY7; S3 Extended Request ID: 61G1/2kXI1/mqclSY6UUjFTzm4ANAtuqMiInnat0VZi/8vYVmyxkZ4GzuTnmjeAcwD02yWHl2eE=) at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:217) at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:151) at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2239) at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2204) at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2143) at ...Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: The bucket is in this region: cn-northwest-1. Please use this region to retry the request (Service: Amazon S3; Status Code: 301; Error Code: 301 Moved Permanently; Request ID: C1AV4REN3P1VGWY7; S3 Extended Request ID: 61G1/2kXI1/mqclSY6UUjFTzm4ANAtuqMiInnat0VZi/8vYVmyxkZ4GzuTnmjeAcwD02yWHl2eE=), S3 Extended Request ID: 61G1/2kXI1/mqclSY6UUjFTzm4ANAtuqMiInnat0VZi/8vYVmyxkZ4GzuTnmjeAcwD02yWHl2eE= at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1640) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1304) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1058) at ...


Set up AWS Credentials and Region for Development - AWS SDK for Java


Get started with the AWS SDK for Java 2.x - AWS SDK for Java


val sc = spark.sparkContext
sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3a.cn-northwest-1.amazonaws.com.cn")


或者


Linux


~/.aws/config[default]
region = cn-northwest-1export AWS_REGION=cn-northwest-1


Windows


C:\Users\USERNAME\.aws\config[default]
region = cn-northwest-1set AWS_REGION=cn-northwest-1


问题3


Exception in thread "main" java.nio.file.AccessDeniedException: s3a://s3-datafacts-poc-001/dct/s3-datafacts-poc-001/dt=2022-05-09: getFileStatus on s3a://s3-datafacts-poc-001/dct/s3-datafacts-poc-001/dt=2022-05-09: com.amazonaws.services.s3.model.AmazonS3Exception: Forbidden (Service: Amazon S3; Status Code: 403; Error Code: 403 Forbidden; Request ID: CB7P31ZHFVCTBZWM; S3 Extended Request ID: Xch4CDPv15SyJQzkaRS6WpJci4o9DR9W2yQcPALY7IgEF/hJutaedBZ3ft5FlaJg8hHrAaPvhNY=), S3 Extended Request ID: Xch4CDPv15SyJQzkaRS6WpJci4o9DR9W2yQcPALY7IgEF/hJutaedBZ3ft5FlaJg8hHrAaPvhNY=:403 Forbidden at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:230) at org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:151) at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2239) at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2204) at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2143) at a:47) ... Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: Forbidden (Service: Amazon S3; Status Code: 403; Error Code: 403 Forbidden; Request ID: CB7P31ZHFVCTBZWM; S3 Extended Request ID: Xch4CDPv15SyJQzkaRS6WpJci4o9DR9W2yQcPALY7IgEF/hJutaedBZ3ft5FlaJg8hHrAaPvhNY=), S3 Extended Request ID: Xch4CDPv15SyJQzkaRS6WpJci4o9DR9W2yQcPALY7IgEF/hJutaedBZ3ft5FlaJg8hHrAaPvhNY= at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1640) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1304) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1058) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:743) at ...


sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3a.cn-northwest-1.amazonaws.com.cn")


问题4


2022-05-10 14:38:15,669 DEBUG org.apache.hadoop.metrics2.impl.MetricsConfig - Could not locate file hadoop-metrics2.properties org.apache.commons.configuration2.ex.ConfigurationException: Could not locate: org.apache.commons.configuration2.io.FileLocator@75de29c0[fileName=hadoop-metrics2.properties,basePath=<null>,sourceURL=,encoding=<null>,fileSystem=<null>,locationStrategy=<null>] at org.apache.commons.configuration2.io.FileLocatorUtils.locateOrThrow(FileLocatorUtils.java:346) at org.apache.commons.configuration2.io.FileHandler.load(FileHandler.java:972) at org.apache.commons.configuration2.io.FileHandler.load(FileHandler.java:702) at org.apache.hadoop.metrics2.impl.MetricsConfig.loadFirst(MetricsConfig.java:116) at org.apache.hadoop.metrics2.impl.MetricsConfig.create(MetricsConfig.java:96) at org.apache.hadoop.metrics2.impl.MetricsSystemImpl.configure(MetricsSystemImpl.java:478) at org.apache.hadoop.metrics2.impl.MetricsSystemImpl.start(MetricsSystemImpl.java:188) at org.apache.hadoop.metrics2.impl.MetricsSystemImpl.init(MetricsSystemImpl.java:163) at org.apache.hadoop.fs.s3a.S3AInstrumentation.getMetricsSystem(S3AInstrumentation.java:253)


在 resources 下创建配置文件 core-site.xml


<configuration><property><name>fs.s3a.aws.credentials.provider</name><value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value></property>
</configuration>


问题5


Caused by: java.lang.UnsupportedOperationException: Unsupported encoding: DELTA_BINARY_PACKED at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.initDataReader(VectorizedColumnReader.java:783) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readPageV2(VectorizedColumnReader.java:833) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.access$100(VectorizedColumnReader.java:54) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader$1.visit(VectorizedColumnReader.java:756) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader$1.visit(VectorizedColumnReader.java:742) at org.apache.parquet.column.page.DataPageV2.accept(DataPageV2.java:141) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readPage(VectorizedColumnReader.java:742) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readBatch(VectorizedColumnReader.java:261) at org.apache.spark.sql.execution.datasources.parquet.VectorizedParquetRecordReader.nextBatch(VectorizedParquetRecordReader.java:283) at org.apache.spark.sql.execution.datasources.parquet.VectorizedParquetRecordReader.nextKeyValue(VectorizedParquetRecordReader.java:181) at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:37) at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:93) at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:173) at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:93) at org.apache.spark.sql.execution.FileSourceScanExec$$anon$1.hasNext(DataSourceScanExec.scala:503) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755) at org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221) at org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299) at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1423) at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1350) at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1414) at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1237) at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:384) at org.apache.spark.rdd.RDD.iterator(RDD.scala:335) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at org.apache.spark.scheduler.Task.run(Task.scala:131) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)


SparkSession = SparkSession.builder().config("spark.sql.parquet.enableVectorizedReader", "false").getOrCreate()


问题6


Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: Bad Request (Service: Amazon S3; Status Code: 400; Error Code: 400 Bad Request; Request ID: Q3SD2NMRY1JHEGHE; S3 Extended Request ID: YW7hSvgBhRgUkInd9coS/BFfnO61SFclCZj4sVYPbY8YYy6zH1n5d8tHFU8I0f1AE7L4YfOZH18=; Proxy: null) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1862) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleServiceErrorResponse(AmazonHttpClient.java:1415) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1384) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1154) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:811) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:779) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:753) at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:713) at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:695) at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:559) at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:539) at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:5437) at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:5384) at com.amazonaws.services.s3.AmazonS3Client.headBucket(AmazonS3Client.java:1445) at com.amazonaws.services.s3.AmazonS3Client.doesBucketExist(AmazonS3Client.java:1381) at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$verifyBucketExists$1(S3AFileSystem.java:381) at org.apache.hadoop.fs.s3a.Invoker.once(Invoker.java:109) ... 95 more


解决


sc.hadoopConfiguration.set("fs.s3a.access.key", "AKIA4ZNT6QH3L45V45VY")
sc.hadoopConfiguration.set("fs.s3a.secret.key", "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")
sc.hadoopConfiguration.set("fs.s3a.endpoint", "s3.cn-northwest-1.amazonaws.com.cn")


问题7


Caused by: java.io.FileNotFoundException: No such file or directory: s3a://s3-datafacts-poc-001/dw/ods/myhudidataset/.hoodie/hoodie.properties at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2310) at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2204) at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2143) at org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:716) at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:906) at org.apache.hudi.common.fs.HoodieWrapperFileSystem.open(HoodieWrapperFileSystem.java:462) at org.apache.hudi.common.config.DFSPropertiesConfiguration.visitFile(DFSPropertiesConfiguration.java:106) ... 91 more


解决


删除 S3 上 hudi 表未成功创建的目录


        <dependency><groupId>org.apache.spark</groupId><artifactId>spark-avro_${scala.binary.version}</artifactId><version>${spark.version}</version><scope>${project.build.scope}</scope></dependency>


问题8


Caused by: java.sql.SQLException: Error while processing statement: FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. org.apache.hadoop.fs.s3a.AWSBadRequestException: doesBucketExist on s3-datafacts-poc-001: com.amazonaws.services.s3.model.AmazonS3Exception: Bad Request (Service: Amazon S3; Status Code: 400; Error Code: 400 Bad Request; Request ID: S38N9ZTM2GHJSAY7; S3 Extended Request ID: gBLBCPfbJjdFYrRPkG6qCEHu3al+LQPr8dQ3TBhocD/E3Cq3Nc5KWZE9ub9vtk7dt1mWJRgQJmc=; Proxy: null), S3 Extended Request ID: gBLBCPfbJjdFYrRPkG6qCEHu3al+LQPr8dQ3TBhocD/E3Cq3Nc5KWZE9ub9vtk7dt1mWJRgQJmc=:400 Bad Request: Bad Request (Service: Amazon S3; Status Code: 400; Error Code: 400 Bad Request; Request ID: S38N9ZTM2GHJSAY7; S3 Extended Request ID: gBLBCPfbJjdFYrRPkG6qCEHu3al+LQPr8dQ3TBhocD/E3Cq3Nc5KWZE9ub9vtk7dt1mWJRgQJmc=; Proxy: null) at org.apache.hive.jdbc.HiveStatement.waitForOperationToComplete(HiveStatement.java:385) at org.apache.hive.jdbc.HiveStatement.execute(HiveStatement.java:254) at org.apache.hudi.hive.ddl.JDBCExecutor.runSQL(JDBCExecutor.java:57) ... 55 more


解决


Hudi同步数据到Hive，无法同步，需要指定hive user


HIVE_USER.key() -> "hive",
HIVE_PASS.key() -> "",


配置协议为 s3


spark.write
..
.save("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table")


问题9


Caused by: java.io.FileNotFoundException: No such file or directory: s3a://s3-datafacts-poc-001/dw/ods/hudi_test_table/.hoodie/.temp/20220513161709/2022-05-13 at org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2310) at org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2204) at org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2143) at org.apache.hadoop.fs.s3a.S3AFileSystem.innerListFiles(S3AFileSystem.java:3148) at org.apache.hadoop.fs.s3a.S3AFileSystem.listFiles(S3AFileSystem.java:3129) at org.apache.hudi.table.marker.DirectWriteMarkers.lambda$createdAndMergedDataPaths$69cdea3b$1(DirectWriteMarkers.java:136) at org.apache.hudi.client.common.HoodieSparkEngineContext.lambda$flatMap$7d470b86$1(HoodieSparkEngineContext.java:78) at org.apache.spark.api.java.JavaRDDLike.$anonfun$flatMap$1(JavaRDDLike.scala:125)


...


原因


本地windows下运行，在s3上创建的目录为s3-datafacts-poc-001/dw/ods/hudi_test_table/.hoodie\.temp/20220513161709/2022-05-13/
windows下写入路径为 \ ，导致路径不正确，无法本地执行。


问题10


Caused by: java.lang.IllegalArgumentException: BlockAlignedAvroParquetWriter does not support scheme s3n at org.apache.hudi.common.fs.HoodieWrapperFileSystem.getHoodieScheme(HoodieWrapperFileSystem.java:159) at org.apache.hudi.common.fs.HoodieWrapperFileSystem.convertToHoodiePath(HoodieWrapperFileSystem.java:132) at org.apache.hudi.io.storage.HoodieParquetWriter.<init>(HoodieParquetWriter.java:56) at org.apache.hudi.io.storage.HoodieFileWriterFactory.newParquetFileWriter(HoodieFileWriterFactory.java:76) at org.apache.hudi.io.storage.HoodieFileWriterFactory.newParquetFileWriter(HoodieFileWriterFactory.java:63) at org.apache.hudi.io.storage.HoodieFileWriterFactory.getFileWriter(HoodieFileWriterFactory.java:49) at org.apache.hudi.io.HoodieCreateHandle.<init>(HoodieCreateHandle.java:101) at org.apache.hudi.io.HoodieCreateHandle.<init>(HoodieCreateHandle.java:74) at org.apache.hudi.io.CreateHandleFactory.create(CreateHandleFactory.java:44) at org.apache.hudi.execution.CopyOnWriteInsertHandler.consumeOneRecord(CopyOnWriteInsertHandler.java:83) at org.apache.hudi.execution.CopyOnWriteInsertHandler.consumeOneRecord(CopyOnWriteInsertHandler.java:40) at org.apache.hudi.common.util.queue.BoundedInMemoryQueueConsumer.consume(BoundedInMemoryQueueConsumer.java:37) at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$2(BoundedInMemoryExecutor.java:121) at java.util.concurrent.FutureTask.run(FutureTask.java:266)


spark.write...
.save("s3://s3-datafacts-poc-001/dw/ods/hudi_test_table")
												


											

Spark3 读写 S3 Parquet, Hive, Hudi相关推荐
	

    
								
  1. Spark SQL与外部数据源的操作（Spark SQL ——＞ CSV/JSON/Parquet/hive/mysql）
		
    目录 一.Spark SQL支持的外部数据源 二.Spark SQL -> CSV 2.1 读CSV文件 a.有列名 b.无列名 2.2 写CSV文件 三.Spark SQL -> JSO ...
    
		
    2. 
						
  2. spark2读取oracle工具类,spark读写Oracle、hive的艰辛之路（一）
		
    前两天工作需求,要通过给的几个Oracle的视图把数据入到hive库中,很遗憾,使用的华为云平台的集区环境中并没有sqoop1,当然也并没有sqoop2,所以,想到的解决方案是使用spark读取Ora ...
    
		
    3. 
						
  3. spark写mysql优化简书_spark读写mysql、hive、kafka数据demo
		
    读取hive库数据 pom.xml依赖配置 org.apache.spark spark-core_2.11 2.1.1 org.apache.spark spark-hive_2.11 2.1.1  ...
    
		
    4. 
						
  4. 使用Apache Hudi + Amazon S3 + Amazon EMR + AWS DMS构建数据湖
		
    1. 引入 数据湖使组织能够在更短的时间内利用多个源的数据,而不同角色用户可以以不同的方式协作和分析数据,从而实现更好.更快的决策.Amazon Simple Storage Service(amaz ...
    
		
    5. 
						
  5. linux怎么看文件是否orc格式,hive文件存储格式orc,parquet,avro对比
		
    orc文件存储格式 ORC文件也是以二进制方式列式存储的,所以是不可以直接读取,ORC文件也是自解析的,它包含许多的元数据,这些元数据都是同构ProtoBuffer进行序列化的.文件结构如下 ORC文 ...
    
		
    6. 
						
  6. parquet to mysql_在hive中使用parquet (CDH4.3)
		
    hadoop版本 cdh4.3 使用impala创建parquet表后,查询会出错. [impala:21000] SELECT * FROM foo;Query: SELECT * FROM foo ...
    
		
    7. 
						
  7. Hive压缩存储（以Parquet为例）
		
    前言 前面说了,HDFS上的数据在生产环境中一定是以压缩格式存储的,那么在Hive中如何使用Parquet格式存储数据呢? 测试数据 在本地服务器上面vim test.txt 1,zs 2,ls 3, ...
    
		
    8. 
						
  8. Hive从入门到放弃——HiveQL表级别DDL设计的艺术性（五）
		
    HiveQL数据库中的表DDL操作   博客Hive从入门到放弃--HiveQL数据库级别DDL设计的艺术性(四)聊完了数据库的基本操作,我们再来聊聊Hive内表的操作. 创建表   官方推荐建表的结 ...
    
		
    9. 
						
  9. hudi的操作记录备份以及文档
		
    Flink datagen代码: public class TestDataKafka2doris {private static final String JDBC_SQL = "CREA ...
    
		
    10. 
		


					

最新文章
	

    
						
  1. 打开闪光灯java代码_android 拍照带水印(可打开闪光灯功能)
		
    2. 
						
  2. Matlab 训练深度学习模型函数 trainingOptions
		
    3. 
						
  3. Mime类型与文件后缀对照表及探测文件MIME的方法
		
    4. 
						
  4. HashMap的实现与优化
		
    5. 
						
  5. 【Hihocoder - offer编程练习赛93 套题题解】交错01串（贪心，暴力）方格矩阵高度（模拟）数对（STLmultiset）修整土地（网络流）
		
    6. 
						
  6. 用PHP制作饼图调查表
		
    7. 
						
  7. 一建管道工程122知识点_过梁、你应该知道的知识点
		
    8. 
						
  8. MySQL无法读表错误的解决方法（MySQL 1018 error）
		
    9. 
						
  9. 进退两难的硅谷程序员们
		
    10. 
						
  10. [Python] L1-015. 跟奥巴马一起画方块-PAT团体程序设计天梯赛GPLT
		
    11. 
						
  11. 这个录屏神器好用哭了！
		
    12. 
						
  12. 计算机专业的核心课程,计算机专业核心课程
		
    13. 
						
  13. Java实现基础坦克大战【含源码】
		
    14. 
						
  14. 上网账号口令怎么获取_怎么从路由器里获取上网帐号和密码
		
    15. 
						
  15. 中小企业信息网络安全解决方案
		
    16. 
						
  16. Linux显示2015年日历表
		
    17. 
						
  17. Google Play商店的各种报错解释以及修复方法。
		
    18. 
						
  18. 手机翻书效果html,移动端实现翻书效果
		
    19. 
						
  19. To Introduce An Anime
		
    20. 
						
  20. 爱签电子合同怎么签，应该注意哪些问题？
		
    21. 
		

	


热门文章
	

    
									
  1. 面向对象的三大特征：封装、继承和多态的简单概述
			
    2. 
						
  2. python 实现表情迁移
			
    3. 
						
  3. 全球首款机场专用自动驾驶纯电动大巴在东京实地测试
			
    4. 
						
  4. CSDN前两百名博客
			
    5. 
						
  5. Angular 个人深究（三）【由InputOutput引起的】
			
    6. 
						
  6. 最新超好看Nteam官网程序源码+全解移除授权
			
    7. 
						
  7. python去中心化_EOS区块链dApp去中心化应用汇总
			
    8. 
						
  8. anbox android 镜像,Anbox将使Ubuntu手机能运行Android应用程序
			
    9. 
						
  9. java 文本文件加密 文本文件解密
			
    10. 
						
  10. 工欲善其事必先利其器–SimpleTestBed
			
    11.