网站防止被采集,wordpress 下载文件,塘厦镇做网站,深圳百度关键字优化撤回流的处理
撤回流是指流式处理过程中#xff0c;两表join过程中的数据是一条一条跑过来的#xff0c;即原本可以join到一起的数据在刚开始可能并没有join上。
撤回流的格式#xff1a; 解决方案 定时器#xff1a;使用定时器定时10s#xff08;数据最大的时间差值两表join过程中的数据是一条一条跑过来的即原本可以join到一起的数据在刚开始可能并没有join上。
撤回流的格式 解决方案 定时器使用定时器定时10s数据最大的时间差值定时器触发时将状态中的数据发送过来如果重复计算这些数据如何保持结果正确即可通过每次度量值修改为当次度量值 - 上次度量值即可
异步IO 减少等待的时间充分利用已有的资源使用异步IO时必须保证从头到尾都是异步的操作即使用异步的连接器
/*** 获取到 redis 的异步连接** return 异步链接对象*/public static StatefulRedisConnectionString, String getRedisAsyncConnection() {RedisClient redisClient RedisClient.create(redis://hadoop102:6379/2);return redisClient.connect();}/*** 关闭 redis 的异步连接** param redisAsyncConn*/public static void closeRedisAsyncConnection(StatefulRedisConnectionString, String redisAsyncConn) {if (redisAsyncConn ! null) {redisAsyncConn.close();}}/*** 获取到 Hbase 的异步连接** return 得到异步连接对象*/public static AsyncConnection getHBaseAsyncConnection() {Configuration conf new Configuration();conf.set(hbase.zookeeper.quorum, hadoop102);conf.set(hbase.zookeeper.property.clientPort, 2181);try {return ConnectionFactory.createAsyncConnection(conf).get();} catch (Exception e) {throw new RuntimeException(e);}}/*** 关闭 hbase 异步连接** param asyncConn 异步连接*/public static void closeAsyncHbaseConnection(AsyncConnection asyncConn) {if (asyncConn ! null) {try {asyncConn.close();} catch (IOException e) {throw new RuntimeException(e);}}}异步IO关联
AsyncDataStream.unorderedWait(异步核心逻辑, 60, TimeUnit.SECONDS) 异步关联维度表CompletableFuture.supplyAsync(new Supplier(){ 异步访问读取Redis中的数据 })返回的数据类型是Future类型 先拼写访问的redisKey获取到dimSkuInfoFuture期货使用dimSkuInfoFuture.get()获取异步结果 thenApplyAsync(new Function()) 旁路缓存判断判断是否在redis中读取到相关数据如果没有读取到需要访问HBase.需要重写HBase的getCells方法改为getAsyncCells方法 连接更换为异步连接Future类型数据需要再get()方法获取具体的值无需关闭连接 将从HBase读取的数据保存到redis, redisAsyncConnection.async().setex(redisKey,24*60*60,dimJsonObj.toJSONString());
异步维度关联封装
继承RichAsyncFunctionTradeSkuOrderBean, TradeSkuOrderBean接口将表名和rowkey拼接的方法抽象化让方法调用者自己传进来封装join方法 join(TradeSkuOrderBean input, JSONOjbect dim); join方法里面填写度量值的聚合逻辑将抽象方法和具体方法分离把抽象方法放到接口中在实现该接口将TradeSkuOrderBean类改为泛型方法T
public abstract class DimAsyncFunctionT extends RichAsyncFunctionT, T
implements DimJoinFunctionT{StatefulRedisConnectionString, String redisAsyncConnection;AsyncConnection hBaseAsyncConnection;String tableName;Overridepublic void open(Configuration parameters) throws Exception {redisAsyncConnection RedisUtil.getRedisAsyncConnection();hBaseAsyncConnection HBaseUtil.getHBaseAsyncConnection();}Overridepublic void close() throws Exception {RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);HBaseUtil.closeAsyncHbaseConnection(hBaseAsyncConnection);}Overridepublic void asyncInvoke(T input, ResultFutureT resultFuture) throws Exception {//java的异步编程方式String tableName getTableName();String rowKey getId(input);String redisKey RedisUtil.getRedisKey(tableName, rowKey);CompletableFuture.supplyAsync(new SupplierString() {Overridepublic String get() {//第一步异步访问得到的数据RedisFutureString dimSkuInfoFuture redisAsyncConnection.async().get(redisKey);String dimInfo null;try {dimInfo dimSkuInfoFuture.get();} catch (Exception e) {e.printStackTrace();}return dimInfo;}}).thenApplyAsync(new FunctionString, JSONObject() {Overridepublic JSONObject apply(String dimInfo) {JSONObject dimJsonObj null;//旁路缓存判断if (dimInfo null || dimInfo.isEmpty()) {try {//需要访问HBasedimJsonObj HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, rowKey);//将读取的数据保存到redisredisAsyncConnection.async().setex(redisKey, 24 * 60 * 60, dimJsonObj.toJSONString());} catch (Exception e) {e.printStackTrace();}} else {//redis中存在缓存数据dimJsonObj JSONObject.parseObject(dimInfo);}return dimJsonObj;}}).thenAccept(new ConsumerJSONObject() {public void accept(JSONObject dim) {//合并维度信息if (dim null) {//无法关联到维度信息System.out.println(无法关联到当前的维度信息: tableName : rowKey);} else {join(input,dim);}//返回结果resultFuture.complete(Collections.singletonList(input));}});}}
