lkness/bi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
1340691923@qq.com 2022-03-02 19:31:17 +08:00
parent 783c289abe
commit ba035974b5
3 changed files with 2 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
cmd/sinker/action/action.go
View File

@ -57,9 +57,6 @@ func MysqlConsumer() {
func AddMetaEvent(kafkaData model.KafkaData) (err error) { func AddMetaEvent(kafkaData model.KafkaData) (err error) {
if kafkaData.ReportType == model.EventReportType { if kafkaData.ReportType == model.EventReportType {
redisConn := db.RedisPool.Get()
defer redisConn.Close()
b := bytes.Buffer{} b := bytes.Buffer{}
b.WriteString(kafkaData.TableId) b.WriteString(kafkaData.TableId)
b.WriteString("_") b.WriteString("_")
@ -128,7 +125,6 @@ func AddTableColumn(kafkaData model.KafkaData, failFunc func(data consumer_data.
} }
} }
b := bytes.Buffer{} b := bytes.Buffer{}
obj.Visit(func(key []byte, v *fastjson.Value) { obj.Visit(func(key []byte, v *fastjson.Value) {
@ -207,7 +203,6 @@ func AddTableColumn(kafkaData model.KafkaData, failFunc func(data consumer_data.
}() }()
}) })
if foundNewKey { if foundNewKey {
dims, err = sinker.ChangeSchema(newKeys, model.GlobConfig.Comm.ClickHouse.DbName, tableName, dims) dims, err = sinker.ChangeSchema(newKeys, model.GlobConfig.Comm.ClickHouse.DbName, tableName, dims)
if err != nil { if err != nil {

2
cmd/sinker/main.go
View File

@ -245,6 +245,7 @@ func main() {
markFn() markFn()
return return
} }
//生成表名 //生成表名
tableName := kafkaData.GetTableName() tableName := kafkaData.GetTableName()
@ -260,6 +261,7 @@ func main() {
return return
} }
//添加元数据 //添加元数据
if err := action.AddMetaEvent(kafkaData); err != nil { if err := action.AddMetaEvent(kafkaData); err != nil {
logs.Logger.Error("addMetaEvent err", zap.Error(err)) logs.Logger.Error("addMetaEvent err", zap.Error(err))

1
platform-basic-libs/sinker/clickhouse.go
View File

@ -50,7 +50,6 @@ func GetDims(database, table string, excludedColumns []string, conn *sqlx.DB) (d
redisConn := db.RedisPool.Get() redisConn := db.RedisPool.Get()
defer redisConn.Close() defer redisConn.Close()
dimsBytes, redisErr := redis.Bytes(redisConn.Do("get", dimsCachekey)) dimsBytes, redisErr := redis.Bytes(redisConn.Do("get", dimsCachekey))
if redisErr == nil && len(dimsBytes) != 0 { if redisErr == nil && len(dimsBytes) != 0 {