云端安全小建议 - 使用 EMR 分析云审计数据

cd /usr/local/service/hive/bin && ./beeline -u "jdbc:hive2://10.0.0.1:7001" -n hadoop -p xxxxxx(你的Hive的密码)

create database cloudaudit location ’cosn://cloudaudit/xxx’;

 CREATE EXTERNAL TABLE `cloudaudit_logs`(            `useridentity` struct<type:string,principalid:string,accountid:string,username:string,secretid:string> COMMENT '账户信息，其中accountid是主账号，principalid是当前操作账号',   `eventtime` string COMMENT '事件发生的时间',      `eventsource` string COMMENT '事件来源',                `sourceipaddress` string COMMENT '事件发起的IP地址',         `eventregion` string COMMENT '资源的地域',       `eventname` string COMMENT '事件名称，例如：RestartInstances',          `resourcetype` string COMMENT '资源类型，例如：CVM',     `useragent` string COMMENT '调用方Agent',              `errorcode` string COMMENT '错误码，0代表正确',       `errormessage` string COMMENT '错误描述',          `additionaleventdata` struct<loginto:string,logintype:string,mfaused:string> COMMENT '登录才有的额外字段',                                `requestid` string COMMENT '请求的唯一ID',      `eventid` string COMMENT '日志的唯一ID',      `eventtype` string COMMENT '请求类型，ConsoleCall代表控制台操作，ApiCall代表API调用，ConsoleLogin代表登录',                              `apiversion` string COMMENT 'api的版本',                    `actiontype` string COMMENT '动作类型',                     `resources` array<string> COMMENT '资源',                 `resourcename` string COMMENT '资源名称')                     PARTITIONED BY (`dt` string) //这个是分区，很重要。目前是一小时一个分区，一个分区对应COS存储桶下的一个目录   ROW FORMAT SERDE                                              'org.openx.data.jsonserde.JsonSerDe'                     WITH SERDEPROPERTIES (                                      'ignore.malformed.json'='true')                        STORED AS INPUTFORMAT                                     'org.apache.hadoop.mapred.TextInputFormat'              OUTPUTFORMAT                                             'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'                   LOCATION                       'cosn://cloudaudit/xxxx'

lastHour = (datetime.datetime.now() - datetime.timedelta(hours=1)).strftime('%Y%m%d%H')hive_client = hiveClient.HiveClient(db_host=config.HIVE_CONFIG['host'], port=config.HIVE_CONFIG['port'],                                        user=config.HIVE_CONFIG['user'], password=config.HIVE_CONFIG['passWord'],                                        database=config.HIVE_CONFIG['database'],                                        authMechanism=config.HIVE_CONFIG['authMechanism'])
# 第一步：首先判断要建的分区是否存在，如果不存在的话新建分区partitionSql = 'show partitions cloudaudit_logs'result = hive_client.query(partitionSql)partitions = []for row in result:    partitions.append(row[0])if 'dt=' + lastHour not in partitions:        alterPartitionSql = 'ALTER TABLE cloudaudit_logs ADD PARTITION (dt="' + lastHour + '") location"cosn://cloudaudit/xxxx/' + lastHour + '"'        hive_client.queryWithoutReturn(alterPartitionSql)

# 第二步：执行数据统计命令sql = 'select useridentity.accountid,userIdentity.principalId,sourceIPAddress,eventName,resourceType,errorCode,eventType,' \          ' count(*) as cnt from cloudaudit_logs where dt = "' + lastHour + '" group by userIdentity.accountid,' \                                                                            'sourceIPAddress,userIdentity.principalId,eventName,resourceType,errorCode,eventType'result = hive_client.query(sql)
hive_client.close()