前言

Flink 官网接入数据源的类型是有限的，不满足自己业务需求需要自己来实现。

Flink还提供了数据源接口（抽象类）,我们实现该接口（继承抽象类）就可以实现自定义数据源，不同的接口（抽象类）功能的丰富性与范围不同，分类如下

SourceFunction: 非并行数据源(并行度只能=1)

RichSourceFunction: 多功能非并行数据源(并行度只能=1)

ParallelSourceFunction: 并行数据源(并行度能够>=1)

RichParallelSourceFunction: 多功能并行数据源(并行度能够>=1)

1.自定义数据源

public class MysqlSource extends RichParallelSourceFunction<Row> {

    PreparedStatement ps = null;
    ResultSet result = null;
    Connection conn = null;

    private boolean flag = true;


    @Override
    public void open(Configuration parameters) throws Exception {
        Connection conn = JdbcUtil.getConn();
        String sql = "SELECT fund_code,underlying_security_id,component_share,creation_redemption_unit,estimate_cash_component," +
                "creation_cash_substitute,substitute_flag,trading_day FROM dim_etf_info";
        ps = conn.prepareStatement(sql);
        super.open(parameters);
    }


    @Override
    public void run(SourceContext<Row> ctx) throws Exception {
        while (flag) {
            result = ps.executeQuery();
            while (result.next()) {
                Row row = new Row(8);
                row.setField(0, result.getString("fund_code"));
                row.setField(1, result.getString("underlying_security_id"));
                row.setField(2, result.getInt("component_share"));
                row.setField(3, result.getBigDecimal("creation_redemption_unit"));
                row.setField(4, result.getBigDecimal("estimate_cash_component"));
                row.setField(5, result.getBigDecimal("creation_cash_substitute"));
                row.setField(6, result.getString("substitute_flag"));
                row.setField(7, result.getDate("trading_day"));
                ctx.collect(row);
            }
            Thread.sleep(1000*60*10);
        }
    }

    @SneakyThrows
    @Override
    public void cancel() {
        flag = false;
    }


    @Override
    public void close() throws Exception {
        JdbcUtil.close(conn, ps, result);
    }
}

2.测试

public class TestCustomSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        DataStreamSource<Row> streamSource = env.addSource(new MysqlSource());
        streamSource.print();
        env.execute("mysql-source");
    }
}

方法描述

open()：数据源最开始打开时执行，整个数据源从加载到销毁，只会执行一次
run(SourceContex)：实现数据获取逻辑，并可以通过传入的参数ctx进行向下游节点的数据转发。
SourceContext：source函数用于发出元素和可能的watermark的接口，确定以及返回source生成的元素的类型。
cancel()：用来取消数据源，一般在run方法中，会存在一个循环来持续产生数据，cancel方法则可以使该循环终止。
close()：数据源关闭时执行，整个数据源从加载到销毁，只会执行一次

sparksql自定义数据源

Spark SQL开放了一系列接入外部数据源的接口，来让开发者可以实现，接口在 org.apache.spark.sql.sources 包下：interfaces.scala 。

（1）DefaultSource

package com.yyds.tags.spark.hbase

import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, DataSourceRegister, RelationProvider}
import org.apache.spark.sql.types.{StringType, StructField, StructType}

/**
 * 默认数据源提供Relation对象，分别为加载数据和保存提供Relation对象
 */
class DefaultSource extends RelationProvider with CreatableRelationProvider with DataSourceRegister{




  val SPERATOR: String = ","
  val HBASE_TABLE_SELECT_FIELDS: String = "selectFields"


  // 使用简称
  override def shortName(): String = {
    "hbase"
  }

  /**
   * 返回BaseRelation实例对象，提供加载数据功能
   * @param sqlContext SQLContext实例对象
   * @param parameters 参数信息
   * @return
   */
  override def createRelation(
                               sqlContext: SQLContext,
                               parameters: Map[String, String]
                             ): BaseRelation = {


      // 1. 定义Schema信息
      val schema: StructType = StructType(
        parameters(HBASE_TABLE_SELECT_FIELDS)
          .split(SPERATOR)
          .map{
            field => StructField(field, StringType, nullable = true)
          })



    // 2. 创建HBaseRelation对象
    val relation = new HbaseRelation(sqlContext, parameters, schema)

    // 3. 返回对象
    relation
  }

  /*** 返回BaseRelation实例对象，提供保存数据功能
   * @param sqlContext SQLContext实例对象
   * @param mode 保存模式
   * @param parameters 参数
   * @param data 数据集
   * @return
   */
  override def createRelation(
                               sqlContext: SQLContext,
                               mode: SaveMode,
                               parameters: Map[String, String],
                               data: DataFrame
                             ): BaseRelation = {

    // 1. 创建HBaseRelation对象
    val relation = new HbaseRelation(sqlContext, parameters, data.schema)

    // 2.保存数据
    relation.insert(data,true)

    // 3.返回
    relation
  }


}

（2）HbaseRelation

package com.yyds.tags.spark.hbase

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.{Put, Result, Scan}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.{TableInputFormat, TableOutputFormat}
import org.apache.hadoop.hbase.protobuf.ProtobufUtil
import org.apache.hadoop.hbase.util.{Base64, Bytes}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, TableScan}
import org.apache.spark.sql.types.StructType

/**
 * 自定义 HBaseRelation 类，继承 BaseRelation 、 TableScan 和 InsertableRelation ，
 * 此外实现序列化接口 Serializable ，所有类声明如下，其中实
 * 现 Serializable 接口为了保证对象可以被序列化和反序列化。
 *
 * 自定义外部数据源：从HBase表加载数据和保存数据值HBase表
 */
case class HbaseRelation(context: SQLContext,
                    params: Map[String, String],
                    userSchema: StructType)
                                   extends BaseRelation
                                   with TableScan
                                   with InsertableRelation
                                   with Serializable {


  // 连接HBase数据库的属性名称
   val HBASE_ZK_QUORUM_KEY: String = "hbase.zookeeper.quorum"
   val HBASE_ZK_QUORUM_VALUE: String = "zkHosts"
   val HBASE_ZK_PORT_KEY: String = "hbase.zookeeper.property.clientPort"
   val HBASE_ZK_PORT_VALUE: String = "zkPort"
   val HBASE_TABLE: String = "hbaseTable"
   val HBASE_TABLE_FAMILY: String = "family"
   val SPERATOR: String = ","
   val HBASE_TABLE_SELECT_FIELDS: String = "selectFields"
   val HBASE_TABLE_ROWKEY_NAME: String = "rowKeyColumn"

  /**
   * sparksql 加载数据和保存程序入口，相当于spark session
   * @return
   */
  override def sqlContext: SQLContext = context

  /**
   * 在sparksql中数据封装在DataFrame或者DataSet中schema信息
   * @return
   */
  override def schema: StructType = userSchema

  /**
   * 从数据源中加载数据，封装在RDD中，每条数据在Row中，和schema信息，转换为DataFrame
   * @return
   */
  override def buildScan(): RDD[Row] = {

    // 读取数据
    /*def newAPIHadoopRDD[K, V, F <: NewInputFormat[K, V]](
               conf: Configuration = hadoopConfiguration,
              fClass: Class[F], kClass: Class[K], vClass: Class[V]
        ): RDD[(K, V)]
   */
    // 1. 读取配置信息，加载HBaseClient配置（主要ZK地址和端口号）
    val conf = HBaseConfiguration.create()
    conf.set(HBASE_ZK_QUORUM_KEY, params(HBASE_ZK_QUORUM_VALUE))
    conf.set(HBASE_ZK_PORT_KEY, params(HBASE_ZK_PORT_VALUE))
    conf.set("zookeeper.znode.parent", "/hbase")

    // 2. 设置表的名称
    conf.set(TableInputFormat.INPUT_TABLE, params(HBASE_TABLE))

    //设置读取列簇和列名称
    val scan: Scan = new Scan()
    scan.addFamily(Bytes.toBytes(params(HBASE_TABLE_FAMILY)))

    val fields = params(HBASE_TABLE_SELECT_FIELDS)
                       .split(SPERATOR)

    fields.foreach{
      field => {
        scan.addColumn(Bytes.toBytes(params(HBASE_TABLE_FAMILY)),Bytes.toBytes(field))
      }
    }

    conf.set(TableInputFormat.SCAN,Base64.encodeBytes(ProtobufUtil.toScan(scan).toByteArray) )


    // 3. 从HBase表加载数据
    val hbaseRDD: RDD[(ImmutableBytesWritable, Result)] =
      sqlContext.sparkContext.newAPIHadoopRDD(conf,
        classOf[TableInputFormat],
        classOf[ImmutableBytesWritable],
        classOf[Result])

    // 将RDD转换为Schema
    // DataFrame = RDD[ROW] + Schema
    val rowsRDD: RDD[Row] = hbaseRDD.map{
      case (_,result) =>
        // 基于列名称获取对应的值
        val values: Seq[String] = fields.map{
          field =>
            val value: Array[Byte] = result.getValue(Bytes.toBytes(params(HBASE_TABLE_FAMILY)),Bytes.toBytes(field))
            // 转换为字符串
            Bytes.toString(value)
        }
        // 将序列转换为Row对象
        Row.fromSeq(values)
    }

    // 返回
    rowsRDD
  }


  /**
   * 将DataFrame保存到数据源
   * @param data 数据集
   * @param overwrite 是否覆写
   */
  override def insert(data: DataFrame, overwrite: Boolean): Unit = {
    // 1. 设置HBase中Zookeeper集群信息
    val conf: Configuration = new Configuration()
    conf.set(HBASE_ZK_QUORUM_KEY, params(HBASE_ZK_QUORUM_VALUE))
    conf.set(HBASE_ZK_PORT_KEY, params(HBASE_ZK_PORT_VALUE))
    // 2. 设置读HBase表的名称
    conf.set(TableOutputFormat.OUTPUT_TABLE, params(HBASE_TABLE))
    // 3. 数据转换
    val columns: Array[String] = data.columns
    val putsRDD: RDD[(ImmutableBytesWritable, Put)] =
      data.rdd.map { row =>
        // 获取RowKey
        val rowKey: String = row.getAs[String](params(HBASE_TABLE_ROWKEY_NAME))
        // 构建Put对象
        val put = new Put(Bytes.toBytes(rowKey))
        // 将每列数据加入Put对象中
        val familyBytes = Bytes.toBytes(params(HBASE_TABLE_FAMILY))
        columns.foreach { column =>
          put.addColumn(
            familyBytes, //
            Bytes.toBytes(column), //
            Bytes.toBytes(row.getAs[String](column)) //
          )
        }
        // 返回二元组
        (new ImmutableBytesWritable(put.getRow), put)
      }
    // 4. 保存数据到表
    putsRDD.saveAsNewAPIHadoopFile(
      s"/apps/hbase/output-" + System.currentTimeMillis(),
      classOf[ImmutableBytesWritable], //
      classOf[Put], //
      classOf[TableOutputFormat[ImmutableBytesWritable]], //
      conf //
    )

  }
}

（3）注册数据源

所以在项目【 resources 】目录下创建库目录【 META-INF/services 】，并且创建文件
【 org.apache.spark.sql.sources.DataSourceRegister 】，内容为数据源主类

com.yyds.tags.spark.hbase.DefaultSource

（4）测试

package com.yyds.tags.hbase.read

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
 * 测试自定义外部数据源实现从HBase表读写数据接口
 */
object HBaseSQLTest {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[4]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()

    // 读取数据
    val usersDF: DataFrame = spark.read
      .format("hbase")
      .option("zkHosts", "192.168.42.7")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_users")
      .option("family", "detail")
      .option("selectFields", "id,gender")
      .load()
    usersDF.printSchema()
    usersDF.cache()
    usersDF.show(10, truncate = false)


    // 保存数据
    usersDF.write
      .mode(SaveMode.Overwrite)
      .format("hbase")
      .option("zkHosts", "192.168.42.7")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_users")
      .option("family", "info")
      .option("rowKeyColumn", "id")
      .save()
    spark.stop()
  }
}

SpringSecurity（二）——自定义数据源

前言

本篇笔记中记录SpringSecurity的认证数据源的自定义，即不在配置文件中修改用户认证的用户名和密码（这种认证是将数据源放入到内存中，运行过程中使用使用内存中的数据源来执行认证），我们将会使用自定义数据源（数据库中的数据）来对用户访问系统资源进行认证。

概述（文字说明，较为晦涩，看不懂的可以直接看下面的代码实现或者找视频，总结部分会有手绘图）

第一篇笔记中并没有说明SpringSecurity的认证流程，在这里给大家说明一下：

首先我们SpringSecurity的认证是由一个叫做AuthenticationManager的接口来完成的，这个接口下面默认有且仅有一个实现类——ProviderManager，认证的具体工作都是由这个实现类来完成。

ProviderManager对象在认证的时候有一个小细节，那就是他会有通过AuthenticationManager接口生成一个具体的实例类来做自己的父类，然而该接口只有一个实现类，即它本身，所以他生成的自己的父类也和自己是同一个类型（典型的我是我爸爸了）。

那么在认证流程中，我们的ProviderManager会先进行认证，如果发现认证失败则会通过自己的父类，即自己生成的同一个类型的实现类按照最基础的配置（默认配置）来进行认证，如果再次认证失败则表示整体认证失败。

在这个过程中，ProviderManager生成的父类其实是一个默认的全局认证器，该认证器是由SpringBoot集成SpringSecurity框架的时候自动注入且按照默认配置设置的一个全局对象（记住这个全局，后面要考），我们首先经过ProviderManager本身而没有限制性父类，代表着我们可以自定义认证规则（具体怎么自定义一会说）。这个时候我们就要引出第三个类——AuthenticationProvider

AuthenticationProvider这个类是一个认证规则，它存在于ProviderManager中，这种情况下叫做自定义认证列表，后期在针对不同类型资源自定义多种不同的认证规则的时候，我们回将所有除默认的认证器以外的所有认证器中进行认证比对，只要有一个通过就表示为认证成功；如果所有自定义的认证规则都没有通过，则会进行默认的认证器规则进行比对，如果默认的认证器规则中比对通过后也表示认证通过。

自定义数据源配置

自定义数据源配置中一般来说可以有两种方式，这两种方式如下：

修改默认认证器对象

在上一篇笔记中我们自定义的配置类中可以自己定义一个方法来修改SpringBoot集成SpringSecurity框架时自动注入的默认认证对象。

/**
 * 该方法为使用SpringBoot默认注入的SpringSecurity提供的AuthenticationManagerBuilder对象来操作
 * 使用这种方式时一定要注意，如果外部重写UserDetailsService 对象的话，一定要将外部重写的该对象注入到这个默认的对象中
 *
 * @throws Exception
 */
@Autowired
public void initialize(AuthenticationManagerBuilder builder) throws Exception {
    log.info("springboot 默认配置：" + builder);
    // 规定用户认证信息存于内存中
    InMemoryUserDetailsManager userDetailsService = new InMemoryUserDetailsManager();
    // 指定用户信息数据
    userDetailsService.createUser(User.withUsername("Reman").password("{noop}123").roles("admin","suser").build());
    // 将自定义的用户数据加入到默认认证器中
    builder.userDetailsService(userDetailsService);
}

在这种自定义的方法中，我们的方法名可以任意命名，只要在内部执行修改默认认证对象的操作即可，同时我们也可以使用最简单的方式来修改。

我们主要需要修改的是用户认证的信息，即：用户名、密码以及权限信息。而这些信息由AuthenticationManagerBuilder实例中的UserDetailsService实例来修改用户的三个信息，所以我们可以通过自定义UserDetailsService配置类来完成修改用户数据，具体如下：

@Bean
public UserDetailsService userDetailsService(){
    InMemoryUserDetailsManager userDetailsService = new InMemoryUserDetailsManager();
    userDetailsService.createUser(User.withUsername("Reman").password("{noop}123").roles("admin","suser").build());
    return userDetailsService;
}

自定义认证器

我们也可以自定义一个认证器对象，然后将之前第定义的用户数据对象注入之中：

/**
 * 推荐使用该方法来进行全局认证管理器的配置
 * 该方法是自定义AuthenticationManagerBuilder对象管理，自定义的对象管理会覆盖SpringBoot默认注入的SpringSecurity对象
 * 这种方式方便我们在项目开发中完全自定义认证管理器
 *
 * @throws Exception
 */
@Override
protected void configure(AuthenticationManagerBuilder builder) throws Exception {
    // 自定义认证管理器时，需要将外部的数据源对象手动注入到对应的自定义的管理其中
    builder.userDetailsService(userDetailsService());
    log.info("自定义AuthenticationManager：" + builder);
}

自定义认证器对象创建的时候属于本地对象，即在工厂中可以使用，但是并不能对外注入。所以我们需要重写一个方法覆盖一下：

// 用来将自定义的AuthenticationManager对象在工厂中进行暴露，使得该对象在项目中的任何地方都可以注入
@Override
// 在工厂中实例一个对象
@Bean
public AuthenticationManager authenticationManagerBean() throws Exception {
    return super.authenticationManagerBean();
}

总结

本篇笔记中记录了自定义认证数据源的一些基本流程和概念。在此重复一下，推荐使用congure方法直接自定义认证器对象覆盖默认注入的对象。

SpringSecurity中的认证流程一定要理解透彻，否则自定义数据源以及后续的一些自定义拦截器等操作会搞混乱，下面手绘图表现出认证器的流程及结构：

上图阐述的是简单的认证的流程，在我们需要认证的时候，首先会在AuthenticationManager接口中实现一个类——ProviderManager，该类在实现后会自行生成一个父类，然而父类是接口，而在默认结构中该接口只有ProviderManager一个实现类，所以会生成一个相同类型的父类。

图中蓝色虚线圈住的部分可以有多个，我们可以根据资源自定义多种规则的认证器。同时一个ProviderManager中可以有多个AuthencationProvider（图中关于最下面的AuthenticationProvider有瑕疵，它应该是一个列表）。

具体认证流程：

• 进入自定义认证器（如果没有则进入默认注入的管理器对象），判断是否符合认证规则；
• 继续进入其他自定义认证器进行判断；
• 符合其中一个认证器的规则，则表示认证通过
• 如果所有自定义认证器都不符合认证规则，则进入父类认证器；
• 父类认证器认证通过则表示认证通过，否则认证失败。