package com.winhc.phoenix.example.job; import com.mongodb.client.MongoDatabase; import com.winhc.phoenix.example.framework.mongo.MongoDbFastScan; import com.winhc.phoenix.example.service.HbaseOperationService; import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.bson.Document; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.stereotype.Component; import java.util.List; import java.util.function.Consumer; import java.util.stream.Collectors; /** * @author: XuJiakai * 2020/10/21 09:52 */ @Slf4j @Component @AllArgsConstructor public class DeleteHbaseByMongoJob { private final MongoTemplate mongoTemplate; private final HbaseOperationService hbaseOperationService; public void start() { MongoDatabase db = mongoTemplate.getDb(); Consumer> func = list -> { List cids = list.stream().map(d -> d.getString("id")).collect(Collectors.toList()); System.out.println(cids); // hbaseOperationService.deleteByRowkey("COMPANY_DYNAMIC", cids); }; MongoDbFastScan mongoDbFastScan = new MongoDbFastScan("xjk_tmp_del_dyn", func, db); // .batchSize(2000).threadNum(5); mongoDbFastScan.scan(); } }