Spark DataFrame 添加列总结

使用kinit 缓存票据 kinit -kt /etc/security/keytabs/hive.service.keytab hive/indata-192-168-44-128.indata.com@INDATA.COM，然后提交Spark程序即可
在spark-submit 中添加参数 –principal hive/indata-192-168-44-128.indata.com@INDATA.COM –keytab /etc/security/keytabs/hive.service.keytab

本地

本地连接，稍微复杂点，首先要配好环境，比如Hadoop的环境变量、winutils等，然后需要配置hosts,将服务器上的/etc/hosts里面的内容拷贝出来，粘贴Windows上的hosts文件里即可

代码

首先需要将集群上的hive-site.xml,core-site.xml,yarn-site.xml,hdfs-site.xml拷贝到src/main/resources文件夹中，其中hive-site.xml是为了连接hive,core-site.xml、hdfs-site.xml和yarn-site.xml是为了认证kerberos

more >>

展开全文 >>

Java 连接 Kerberos认证下的Spark Thrift Server/Hive Server总结

2021-05-10

前言

总结Java如何连接Kerberos认证下的Spark Thrift Server/Hive Server总结

启动

关于如何启动 Spark Thrift Server和 Hive Server 请参考https://dongkelun.com/2021/02/19/javaSparkThriftServer/

Java 代码

pom 依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-jdbc</artifactId>
        <version>1.2.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.6.5</version>
    </dependency>
</dependencies>

关于依赖版本对应同样请参考https://dongkelun.com/2021/02/19/javaSparkThriftServer/

more >>

展开全文 >>

Spark覆盖写入mysql表但不改变已有的表结构

2021-04-29

前言

本文记录Spark如何在表存在的情况时覆盖写入mysql但不修改已有的表结构，并进行主要的源码跟踪以了解其实现原理。主要场景为先用建表语句建好mysql表，然后用spark导入数据，可能会存在多次全表覆写导入的情况。

代码

已上传github

主要的参数为.option(“truncate”, true)，可以参考Spark官网http://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

主要代码逻辑为，读取csv，进行日期转化，然后覆盖写入到已经建好的mysql表中。

package com.dkl.blog.spark.mysql

import java.util.Properties

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, to_date}

/**
 * Created by dongkelun on 2021/4/29 14:06
 *
 * Spark覆盖写入mysql表但不修改已有的表结构
 */
object SparkMysqlOverwriteTruncateTable {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("SparkMysqlOverwriteTruncateTable")
      .master("local[*]")
      .getOrCreate()

    //rewriteBatchedStatements参数为批量写入数据，可以增加写入效率
    val url = "jdbc:mysql://192.168.44.128:3306/test?useUnicode=true&characterEncoding=utf-8&rewriteBatchedStatements=true"
    val tableName = "trafficbase_cljbxx"
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "Root-123456")

    //读取本地csv
    var df = spark.read.option("header", "true").csv("D:\\文档\\inspur\\csg\\功能测评\\测试数据\\trafficbase_cljbxx.csv")
    //字符串转为日期类型
    df = df.withColumn("czsj", to_date(col("czsj"), "dd/mm/yyyy"))
    df.show()

    df.write.mode("overwrite")
      .option("truncate", true) //覆盖写入数据前先truncate table而不是drop table
      .jdbc(url = url, table = tableName, prop)
    spark.stop
  }
}

more >>

展开全文 >>