conn

Documentation

Overview

Package conn @author: Violet-Eva @date : 2024/12/24 @notes :

Package conn @author: Violet-Eva @date : 2024/12/19 @notes :

Package conn @author: Violet-Eva @date : 2024/12/26 @notes :

Package conn @author: Violet-Eva @date : 2024/12/24 @notes :

Package conn @author: Violet-Eva @date : 2024/12/26 @notes :

Package conn @author: Violet-Eva @date : 2024/12/24 @notes :

Index

• Constants
• func InitGormDB(dbType, dbName, user, passwd, host string, ...) (gormDB *gorm.DB, err error)
• func SAToTSA(structType *types.StructType, data ...any) [][]interface{}
• func StructToStructType(v interface{}, isRename bool) (*types.StructType, error)
• type Address
• type HiveConn
  • func NewHiveConn(retryCount int, retryInterval time.Duration, queryTimeout int, ...) *HiveConn
  • func (hc *HiveConn) Close() (err error)
  • func (hc *HiveConn) ExecQuery(query string) ([]map[string]interface{}, error)
  • func (hc *HiveConn) ExecQueryBatchSize(query string, batchSize int, ...) (err error)
  • func (hc *HiveConn) ExecQueryNoResult(query string) error
  • func (hc *HiveConn) ExecQueryToStruct(query string, data any) error
  • func (hc *HiveConn) GetHiveConn() (err error)
• type HiveConnInformation
  • func NewHiveConnInformation(addresses []Address, auth string, service string, fetchSize int64, ...) *HiveConnInformation
• type KrbAuth
  • func NewKrbAuth(krbConfPath, kinitProgramPath, keyTabFilePath, principal string) *KrbAuth
  • func (ka *KrbAuth) Kinit() error
• type PgDB
  • func InitPGConn(user, passwd, host, dbname string, ...) (pdb *PgDB, err error)
  • func (p *PgDB) ExecQuery(query string, args ...interface{}) (list []map[string]interface{}, err error)
  • func (p *PgDB) ExecQueryNoResult(query string, args ...interface{}) (err error)
  • func (p *PgDB) ExecQueryToString(query string, args ...interface{}) (list []map[string]string, err error)
  • func (p *PgDB) ExecQueryToStruct(query string, data any, args ...interface{}) (err error)
• type SQLDB
  • func InitSQLDB(dbType, dbName, user, passwd, host string, ...) (sqlDB *SQLDB, err error)
  • func (s *SQLDB) Close() (err error)
  • func (s *SQLDB) ExecQuery(query string, args ...interface{}) (list []map[string]interface{}, err error)
  • func (s *SQLDB) ExecQueryBatchProcessing(query string, batchSize int, ...) (err error)
  • func (s *SQLDB) ExecQueryNoResult(query string, args ...interface{}) (err error)
  • func (s *SQLDB) ExecQueryToString(query string, args ...interface{}) (list []map[string]string, err error)
  • func (s *SQLDB) ExecQueryToStruct(query string, data any, args ...interface{}) (err error)
• type SparkSQL
  • func NewSparkSQL(ip string, port int, args ...map[string]string) (*SparkSQL, error)
  • func (s *SparkSQL) CreateDataFrame(ctx context.Context, data [][]any, schema *types.StructType) (sql.DataFrame, error)
  • func (s *SparkSQL) CreateDataFrameFromMap(v interface{}, isTag bool, isRename bool, data ...map[string]interface{}) (sql.DataFrame, error)
  • func (s *SparkSQL) CreateDataFrameFromStruct(data any, isRename bool) (sql.DataFrame, error)
  • func (s *SparkSQL) DFCollectBatchProcessingForInterface(df sql.DataFrame, batchSize int, ...) (err error)
  • func (s *SparkSQL) DFCollectBatchProcessingForString(df sql.DataFrame, batchSize int, ...) (err error)
  • func (s *SparkSQL) Exec(query string) (sql.DataFrame, error)
  • func (s *SparkSQL) ExecQuery(query string) (output []map[string]interface{}, err error)
  • func (s *SparkSQL) ExecQueryBatchProcessingForInterface(query string, batchSize int, ...) (err error)
  • func (s *SparkSQL) ExecQueryBatchProcessingForString(query string, batchSize int, function ...func(input []map[string]string) error) (err error)
  • func (s *SparkSQL) ExecQueryToMapString(query string) (output []map[string]string, err error)

Constants

const (
	DefaultKrbConfPath  = "/etc/krb5.conf"
	DefaultKinitProgram = "/usr/bin/kinit"
)

Functions

func InitGormDB

func InitGormDB(dbType, dbName, user, passwd, host string, port, retryCount, retryInterval, maxIdleConn, maxOpenConn, maxLifetime int, opts ...string) (gormDB *gorm.DB, err error)

func SAToTSA

func SAToTSA(structType *types.StructType, data ...any) [][]interface{}

SAToTSA @Description: any slice -> any 2D slicing @param structType @param data @return [][]interface{}

func StructToStructType

func StructToStructType(v interface{}, isRename bool) (*types.StructType, error)

StructToStructType @Description: isTag is false , get struct elem name assign to structField name. isTag is true , get json tag name assign to structField name. @param v @param isTag @return *types.StructType @return error

Types

type Address

type Address struct {
	Host string
	Port int
	// binary http
	Mode string
}

type HiveConn

type HiveConn struct {
	KA            *KrbAuth
	HCI           *HiveConnInformation
	Conn          *gohive.Connection
	RetryCount    int
	RetryInterval time.Duration
	QueryTimeout  int
}

func NewHiveConn

func NewHiveConn(retryCount int, retryInterval time.Duration, queryTimeout int, information *HiveConnInformation, auth *KrbAuth) *HiveConn

func (*HiveConn) Close

func (hc *HiveConn) Close() (err error)

func (*HiveConn) ExecQuery

func (hc *HiveConn) ExecQuery(query string) ([]map[string]interface{}, error)

func (*HiveConn) ExecQueryBatchSize

func (hc *HiveConn) ExecQueryBatchSize(query string, batchSize int, function ...func(input []map[string]interface{}) error) (err error)

func (*HiveConn) ExecQueryNoResult

func (hc *HiveConn) ExecQueryNoResult(query string) error

func (*HiveConn) ExecQueryToStruct

func (hc *HiveConn) ExecQueryToStruct(query string, data any) error

ExecQueryToStruct @Description: @param query @param data pointer @return []map[string]interface{} @return error

func (*HiveConn) GetHiveConn

func (hc *HiveConn) GetHiveConn() (err error)

type HiveConnInformation

type HiveConnInformation struct {
	Addresses []Address
	// KERBEROS NONE NOSASL LDAP CUSTOM DIGEST-MD5
	Auth          string
	Configuration *gohive.ConnectConfiguration
	ConnTimeout   time.Duration
}

func NewHiveConnInformation

func NewHiveConnInformation(addresses []Address, auth string, service string, fetchSize int64, hiveConfig map[string]string, connTimeout time.Duration) *HiveConnInformation

type KrbAuth

type KrbAuth struct {
	KrbConfPath      string
	KinitProgramPath string
	KeyTabFilePath   string
	Principal        string
}

func NewKrbAuth

func NewKrbAuth(krbConfPath, kinitProgramPath, keyTabFilePath, principal string) *KrbAuth

func (*KrbAuth) Kinit

func (ka *KrbAuth) Kinit() error

type PgDB

type PgDB struct {
	PostgresDB   *pgxpool.Conn
	QueryTimeout int
}

func InitPGConn

func InitPGConn(user, passwd, host, dbname string, port, retryCount, retryInterval, queryTimeout int, opts ...string) (pdb *PgDB, err error)

func (*PgDB) ExecQuery

func (p *PgDB) ExecQuery(query string, args ...interface{}) (list []map[string]interface{}, err error)

func (*PgDB) ExecQueryNoResult

func (p *PgDB) ExecQueryNoResult(query string, args ...interface{}) (err error)

func (*PgDB) ExecQueryToString

func (p *PgDB) ExecQueryToString(query string, args ...interface{}) (list []map[string]string, err error)

func (*PgDB) ExecQueryToStruct

func (p *PgDB) ExecQueryToStruct(query string, data any, args ...interface{}) (err error)

type SQLDB

type SQLDB struct {
	QueryTimeOut int
	SQLDB        *sql.DB
}

func InitSQLDB

func InitSQLDB(dbType, dbName, user, passwd, host string, port, retryCount, retryInterval, queryTimeout, maxIdleConn, maxOpenConn, maxLifetime int, opts ...string) (sqlDB *SQLDB, err error)

func (*SQLDB) Close

func (s *SQLDB) Close() (err error)

func (*SQLDB) ExecQuery

func (s *SQLDB) ExecQuery(query string, args ...interface{}) (list []map[string]interface{}, err error)

func (*SQLDB) ExecQueryBatchProcessing

func (s *SQLDB) ExecQueryBatchProcessing(query string, batchSize int, function ...func(input []map[string]interface{}) error) (err error)

func (*SQLDB) ExecQueryNoResult

func (s *SQLDB) ExecQueryNoResult(query string, args ...interface{}) (err error)

func (*SQLDB) ExecQueryToString

func (s *SQLDB) ExecQueryToString(query string, args ...interface{}) (list []map[string]string, err error)

func (*SQLDB) ExecQueryToStruct

func (s *SQLDB) ExecQueryToStruct(query string, data any, args ...interface{}) (err error)

type SparkSQL

type SparkSQL struct {
	sql.SparkSession
	// contains filtered or unexported fields
}

func NewSparkSQL

func NewSparkSQL(ip string, port int, args ...map[string]string) (*SparkSQL, error)

func (*SparkSQL) CreateDataFrame

func (s *SparkSQL) CreateDataFrame(ctx context.Context, data [][]any, schema *types.StructType) (sql.DataFrame, error)

func (*SparkSQL) CreateDataFrameFromMap

func (s *SparkSQL) CreateDataFrameFromMap(v interface{}, isTag bool, isRename bool, data ...map[string]interface{}) (sql.DataFrame, error)

CreateDataFrameFromMap @Description: []map[string]interface{} -> sql.DataFrame , map @param ctx @param v @param isTag true ,Assign values to dataframes based on JSON tags. false, Assign values to dataframes based on elem name. This field is because it is not possible to ignore the case match to the value, and the field is added. @param isRename true ,Rename the dataframe based on the spark tag , Insufficient tags are supplemented by elem name @param data @return sql.DataFrame @return error

func (*SparkSQL) CreateDataFrameFromStruct

func (s *SparkSQL) CreateDataFrameFromStruct(data any, isRename bool) (sql.DataFrame, error)

CreateDataFrameFromStruct @Description: @param ctx @param data @param isRename == true ,Rename the dataframe based on the spark tag , Insufficient tags are supplemented by elem name @return sql.DataFrame @return error

func (*SparkSQL) DFCollectBatchProcessingForInterface

func (s *SparkSQL) DFCollectBatchProcessingForInterface(df sql.DataFrame, batchSize int, function ...func(input []map[string]interface{}) error) (err error)

func (*SparkSQL) DFCollectBatchProcessingForString

func (s *SparkSQL) DFCollectBatchProcessingForString(df sql.DataFrame, batchSize int, function ...func(input []map[string]string) error) (err error)

func (*SparkSQL) Exec

func (s *SparkSQL) Exec(query string) (sql.DataFrame, error)

func (*SparkSQL) ExecQuery

func (s *SparkSQL) ExecQuery(query string) (output []map[string]interface{}, err error)

func (*SparkSQL) ExecQueryBatchProcessingForInterface

func (s *SparkSQL) ExecQueryBatchProcessingForInterface(query string, batchSize int, function ...func(input []map[string]interface{}) error) (err error)

func (*SparkSQL) ExecQueryBatchProcessingForString

func (s *SparkSQL) ExecQueryBatchProcessingForString(query string, batchSize int, function ...func(input []map[string]string) error) (err error)

func (*SparkSQL) ExecQueryToMapString

func (s *SparkSQL) ExecQueryToMapString(query string) (output []map[string]string, err error)