大数据架构核心六层体系

// DataX核心执行逻辑伪代码
public class DataXEngine {
    private List<Plugin> readerPlugins;  // 数据读取插件
    private List<Plugin> writerPlugins; // 数据写入插件
    
    public void executeJob(JobConfig config) {
        // 1. 初始化读写插件
        Reader reader = initReader(config.getReaderConfig());
        Writer writer = initWriter(config.getWriterConfig());
        
        // 2. 构建数据传输通道
        Channel channel = createChannel(config.getChannelConfig());
        
        // 3. 并发执行数据同步
        List<Thread> threads = createThreadPool(reader, writer, channel);
        waitForCompletion(threads);
        
        // 4. 统计与错误处理
        reportMetrics(calculateMetrics());
    }
}

// MySQL CDC源定义
DebeziumSourceFunction<String> source = MySQLSource.<String>builder()
    .hostname("mysql-host")
    .port(3306)
    .databaseList("test_db")
    .tableList("test_db.orders")
    .username("flinkuser")
    .password("flinkpw")
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

// 流处理执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(source).print();

// HDFS写数据流程核心逻辑
public class HDFSWriteProcess {
    public void writeFile(FSDataOutputStream stream, byte[] data) throws IOException {
        // 1. 数据分包（默认128MB块大小）
        List<Packet> packets = splitIntoPackets(data, 128 * 1024 * 1024);
        
        // 2. 管道化写入（DataNode副本复制）
        Pipeline pipeline = createPipeline(3); // 3副本配置
        
        for (Packet packet : packets) {
            // 3. 校验和计算
            byte[] checksum = calculateChecksum(packet);
            
            // 4. 顺序写入数据节点
            pipeline.write(packet, checksum);
        }
        
        // 5. 关闭流并确认写入
        pipeline.close();
        stream.close();
    }
}

// HBase Java API操作示例
Configuration config = HBaseConfiguration.create();
config.set("hbase.zookeeper.quorum", "zk1,zk2,zk3");

try (Connection connection = ConnectionFactory.createConnection(config);
     Table table = connection.getTable(TableName.valueOf("user"))) {
    
    // 写入数据
    Put put = new Put(Bytes.toBytes("rowkey1"));
    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("name"), Bytes.toBytes("Alice"));
    put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("age"), Bytes.toBytes("25"));
    table.put(put);
    
    // 读取数据
    Get get = new Get(Bytes.toBytes("rowkey1"));
    Result result = table.get(get);
    byte[] name = result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("name"));
    System.out.println("Name: " + Bytes.toString(name));
    
    // 范围扫描
    Scan scan = new Scan();
    scan.setStartRow(Bytes.toBytes("row100"));
    scan.setStopRow(Bytes.toBytes("row200"));
    ResultScanner scanner = table.getScanner(scan);
    for (Result res : scanner) {
        // 处理结果
    }
}

// Flink实时ETL示例
public class RealTimeETLJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);  // 5秒一次Checkpoint
        env.setParallelism(4);
        
        // 定义Kafka源
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("user_events")
            .setGroupId("flink_etl")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
            
        // 实时数据处理流水线
        DataStream<UserEvent> events = env.fromSource(
                source, WatermarkStrategy.noWatermarks(), "Kafka Source")
            .map(new JsonToUserEventMapper())  // JSON解析
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
            )  // 水印生成
            .filter(event -> event.isValid())  // 数据过滤
            .keyBy(UserEvent::getUserId);  // 按键分区
        
        // 窗口聚合计算
        DataStream<UserBehaviorSummary> summary = events
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new UserBehaviorAggregator());
            
        // 输出到多种目的地
        summary.addSink(new KafkaSink<>("behavior_summary"));
        summary.addSink(new ElasticsearchSink<>("user_behavior_index"));
        
        env.execute("Real-time User Behavior ETL");
    }
}

// 自定义聚合函数
public class UserBehaviorAggregator implements AggregateFunction<UserEvent, Accumulator, UserBehaviorSummary> {
    @Override
    public Accumulator createAccumulator() {
        return new Accumulator();
    }
    
    @Override
    public Accumulator add(UserEvent event, Accumulator accumulator) {
        accumulator.addEvent(event);
        return accumulator;
    }
    
    @Override
    public UserBehaviorSummary getResult(Accumulator accumulator) {
        return accumulator.getSummary();
    }
    
    @Override
    public Accumulator merge(Accumulator a, Accumulator b) {
        return a.merge(b);
    }
}

// 基于Flink CEP的复杂事件处理
Pattern<UserEvent, ?> riskPattern = Pattern.<UserEvent>begin("first")
    .where(new SimpleCondition<UserEvent>() {
        @Override
        public boolean filter(UserEvent event) {
            return event.getType().equals("login");
        }
    })
    .next("second")
    .where(new SimpleCondition<UserEvent>() {
        @Override
        public boolean filter(UserEvent event) {
            return event.getType().equals("password_change");
        }
    })
    .within(Time.minutes(5));  // 5分钟内连续发生

CEP.pattern(events.keyBy(UserEvent::getUserId), riskPattern)
    .process(new PatternProcessFunction<UserEvent, RiskAlert>() {
        @Override
        public void processMatch(Map<String, List<UserEvent>> match, 
                                Context ctx, 
                                Collector<RiskAlert> out) {
            UserEvent first = match.get("first").get(0);
            UserEvent second = match.get("second").get(0);
            
            RiskAlert alert = new RiskAlert(
                first.getUserId(),
                "SUSPICIOUS_BEHAVIOR",
                "Login followed immediately by password change"
            );
            out.collect(alert);
        }
    });

// 同一代码处理批和流数据
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 定义数据源（批或流）
DataStream<Order> orders = env.fromSource(
    isStreaming ? createKafkaSource() : createFileSource(),
    WatermarkStrategy.forMonotonousTimestamps(),
    isStreaming ? "Kafka Orders" : "File Orders"
);

// 统一数据处理逻辑
DataStream<OrderSummary> summary = orders
    .keyBy(Order::getProductId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .aggregate(new OrderAggregator());

// 统一输出
summary.sinkTo(isStreaming ? createKafkaSink() : createFileSink());

// Spring Boot数据API示例
@RestController
@RequestMapping("/api/data")
public class DataApiController {
    
    @Autowired
    private DataService dataService;
    
    // 分页查询接口
    @GetMapping("/sales")
    public ResponseEntity<PageResult<SaleRecord>> querySales(
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate startDate,
            @RequestParam @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate endDate,
            @RequestParam(required = false) String region,
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "50") int size) {
        
        // 构建查询条件
        SalesQuery query = SalesQuery.builder()
            .startDate(startDate)
            .endDate(endDate)
            .region(region)
            .page(page)
            .size(size)
            .build();
        
        // 执行查询
        PageResult<SaleRecord> result = dataService.querySales(query);
        
        return ResponseEntity.ok()
            .header("X-Total-Count", String.valueOf(result.getTotal()))
            .body(result);
    }
    
    // 实时数据推送接口
    @GetMapping(value = "/sales/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<SaleRecord> streamSales(
            @RequestParam String region) {
        return dataService.getSalesStream(region)
            .delayElements(Duration.ofSeconds(1))
            .onBackpressureBuffer(1000);
    }
    
    // 数据写入接口
    @PostMapping("/sales")
    public ResponseEntity<SaleRecord> createSale(@RequestBody SaleRecord record) {
        SaleRecord created = dataService.createSale(record);
        return ResponseEntity.created(URI.create("/api/data/sales/" + created.getId()))
                .body(created);
    }
}

// 基于Resilience4j的服务熔断
@Slf4j
@Service
public class DataQueryService {
    
    @Autowired
    private DataSource dataSource;
    
    // 熔断器配置：10秒内50%失败率触发熔断
    @CircuitBreaker(name = "dataQuery", fallbackMethod = "fallbackQuery")
    public List<DataRecord> queryData(DataQuery query) {
        return dataSource.executeQuery(query);
    }
    
    // 降级方法
    private List<DataRecord> fallbackQuery(DataQuery query, Throwable t) {
        log.warn("Data query circuit open, using fallback", t);
        
        // 返回缓存数据或空结果
        return Collections.emptyList();
    }
    
    // 限流配置：每秒10个请求
    @RateLimiter(name = "dataQuery", fallbackMethod = "rateLimitFallback")
    public List<DataRecord> rateLimitedQuery(DataQuery query) {
        return dataSource.executeQuery(query);
    }
    
    private List<DataRecord> rateLimitFallback(DataQuery query) {
        throw new TooManyRequestsException("Rate limit exceeded");
    }
}

// Spring Security数据权限控制
@Configuration
@EnableWebSecurity
public class DataSecurityConfig extends WebSecurityConfigurerAdapter {
    
    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http
            .authorizeRequests()
                .antMatchers("/api/public/**").permitAll()
                .antMatchers("/api/data/**").hasRole("DATA_USER")
                .antMatchers(HttpMethod.GET, "/api/data/query").hasAuthority("DATA_READ")
                .antMatchers(HttpMethod.POST, "/api/data/**").hasAuthority("DATA_WRITE")
                .anyRequest().authenticated()
            .and()
            .oauth2ResourceServer()
                .jwt()
                .jwtAuthenticationConverter(new DataJwtConverter());
    }
}

// 数据行级权限控制
@Component
public class DataRowSecurityService {
    
    public Predicate<DataRecord> getRowFilter(User user) {
        // 基于用户角色和数据标签的权限过滤
        return record -> {
            if (user.hasRole("ADMIN")) {
                return true;
            }
            
            // 部门数据隔离
            if (user.getDepartment() != null) {
                return user.getDepartment().equals(record.getDepartment());
            }
            
            // 区域数据隔离
            if (user.getRegion() != null) {
                return user.getRegion().equals(record.getRegion());
            }
            
            return false;
        };
    }
}

// 敏感数据脱敏处理
public class DataMaskingService {
    
    private static final String MASK_CHAR = "*";
    
    public String maskEmail(String email) {
        if (email == null) return null;
        
        int atIndex = email.indexOf("@");
        if (atIndex <= 1) return email;
        
        String name = email.substring(0, atIndex);
        String domain = email.substring(atIndex);
        
        return name.charAt(0) + MASK_CHAR.repeat(3) + domain;
    }
    
    public String maskPhone(String phone) {
        if (phone == null) return null;
        
        if (phone.length() <= 4) return phone;
        
        return MASK_CHAR.repeat(phone.length() - 4) 
             + phone.substring(phone.length() - 4);
    }
    
    public String maskIdCard(String idCard) {
        if (idCard == null) return null;
        
        if (idCard.length() <= 8) return idCard;
        
        return idCard.substring(0, 4) 
             + MASK_CHAR.repeat(idCard.length() - 8) 
             + idCard.substring(idCard.length() - 4);
    }
}

// 数据加密服务
@Service
public class DataEncryptionService {
    
    @Value("${encryption.key}")
    private String encryptionKey;
    
    public String encryptData(String data) {
        try {
            Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
            SecretKey key = new SecretKeySpec(encryptionKey.getBytes(), "AES");
            cipher.init(Cipher.ENCRYPT_MODE, key);
            
            byte[] encrypted = cipher.doFinal(data.getBytes(StandardCharsets.UTF_8));
            return Base64.getEncoder().encodeToString(encrypted);
        } catch (Exception e) {
            throw new RuntimeException("Data encryption failed", e);
        }
    }
    
    public String decryptData(String encryptedData) {
        try {
            Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
            SecretKey key = new SecretKeySpec(encryptionKey.getBytes(), "AES");
            cipher.init(Cipher.DECRYPT_MODE, key);
            
            byte[] decoded = Base64.getDecoder().decode(encryptedData);
            byte[] decrypted = cipher.doFinal(decoded);
            return new String(decrypted, StandardCharsets.UTF_8);
        } catch (Exception e) {
            throw new RuntimeException("Data decryption failed", e);
        }
    }
}

// Atlas元数据采集示例
public class AtlasMetadataCollector {
    
    private AtlasClientV2 atlasClient;
    
    public void collectHiveMetadata(Table table) {
        // 构建表实体
        AtlasEntity tableEntity = new AtlasEntity(HIVE_TABLE_TYPE);
        tableEntity.setAttribute("name", table.getName());
        tableEntity.setAttribute("db", table.getDbName());
        tableEntity.setAttribute("owner", table.getOwner());
        tableEntity.setAttribute("createTime", table.getCreateTime());
        
        // 构建列实体
        List<AtlasEntity> columnEntities = new ArrayList<>();
        for (Column column : table.getColumns()) {
            AtlasEntity columnEntity = new AtlasEntity(HIVE_COLUMN_TYPE);
            columnEntity.setAttribute("name", column.getName());
            columnEntity.setAttribute("type", column.getType());
            columnEntity.setAttribute("comment", column.getComment());
            columnEntities.add(columnEntity);
        }
        
        // 建立关联关系
        AtlasRelationship tableColumnRel = new AtlasRelationship(
            HIVE_TABLE_COLUMNS_RELATIONSHIP_TYPE);
        tableColumnRel.setAttribute("table", tableEntity);
        tableColumnRel.setAttribute("columns", columnEntities);
        
        // 提交到Atlas
        atlasClient.createEntities(Arrays.asList(tableEntity));
        atlasClient.createRelationship(tableColumnRel);
    }
    
    public void trackDataLineage(Process process) {
        // 构建数据处理血缘
        AtlasProcess processEntity = new AtlasProcess(DATA_PROCESS_TYPE);
        processEntity.setAttribute("name", process.getName());
        processEntity.setAttribute("type", process.getType());
        
        // 输入输出关系
        for (Dataset input : process.getInputs()) {
            AtlasRelationship inputRel = new AtlasRelationship(
                PROCESS_INPUTS_RELATIONSHIP_TYPE);
            inputRel.setAttribute("process", processEntity);
            inputRel.setAttribute("input", toAtlasEntity(input));
            atlasClient.createRelationship(inputRel);
        }
        
        for (Dataset output : process.getOutputs()) {
            AtlasRelationship outputRel = new AtlasRelationship(
                PROCESS_OUTPUTS_RELATIONSHIP_TYPE);
            outputRel.setAttribute("process", processEntity);
            outputRel.setAttribute("output", toAtlasEntity(output));
            atlasClient.createRelationship(outputRel);
        }
    }
}

// 数据质量规则定义与执行
public class DataQualityEngine {
    
    private List<QualityRule> rules;
    
    public QualityResult validateDataset(Dataset dataset) {
        QualityResult result = new QualityResult();
        
        for (QualityRule rule : rules) {
            RuleResult ruleResult = rule.validate(dataset);
            result.addRuleResult(ruleResult);
            
            if (ruleResult.getStatus() == RuleStatus.ERROR 
                && rule.isBlocking()) {
                break;  // 阻塞性规则失败立即终止
            }
        }
        
        return result;
    }
}

// 数据质量规则接口
public interface QualityRule {
    RuleResult validate(Dataset dataset);
    boolean isBlocking();
    String getDescription();
}

// 具体规则实现
public class CompletenessRule implements QualityRule {
    
    private String columnName;
    private double threshold;  // 完整率阈值
    
    @Override
    public RuleResult validate(Dataset dataset) {
        long totalCount = dataset.count();
        long nonNullCount = dataset.filter(col(columnName).isNotNull()).count();
        double completeness = (double) nonNullCount / totalCount;
        
        RuleStatus status = completeness >= threshold ? 
            RuleStatus.PASS : RuleStatus.ERROR;
        
        return new RuleResult(status, 
            String.format("Completeness: %.2f%%", completeness * 100));
    }
}

public class ValueRangeRule implements QualityRule {
    
    private String columnName;
    private double minValue;
    private double maxValue;
    
    @Override
    public RuleResult validate(Dataset dataset) {
        long outOfRangeCount = dataset.filter(
            col(columnName).lt(minValue).or(col(columnName).gt(maxValue))
        ).count();
        
        double violationRate = (double) outOfRangeCount / dataset.count();
        RuleStatus status = violationRate <= 0.01 ?  // 允许1%的异常
            RuleStatus.PASS : RuleStatus.WARNING;
        
        return new RuleResult(status,
            String.format("Violation rate: %.2f%%", violationRate * 100));
    }
}

// 数据敏感度分级
public enum DataSensitivity {
    PUBLIC,          // 公开信息
    INTERNAL,        // 内部信息
    CONFIDENTIAL,    // 机密信息
    HIGHLY_CONFIDENTIAL, // 高度机密
    RESTRICTED       // 受限信息
}

// 数据分类服务
@Service
public class DataClassificationService {
    
    private Map<Pattern, DataSensitivity> patternRules;
    
    public DataSensitivity classifyData(String data, String context) {
        // 基于正则表达式匹配的敏感数据识别
        for (Map.Entry<Pattern, DataSensitivity> entry : patternRules.entrySet()) {
            if (entry.getKey().matcher(data).find()) {
                return entry.getValue();
            }
        }
        
        // 基于上下文环境的分类
        if (context.contains("financial") || context.contains("payment")) {
            return DataSensitivity.CONFIDENTIAL;
        }
        
        if (context.contains("public") || context.contains("marketing")) {
            return DataSensitivity.PUBLIC;
        }
        
        return DataSensitivity.INTERNAL;
    }
    
    public Map<String, DataSensitivity> classifyDataset(Dataset dataset) {
        Map<String, DataSensitivity> result = new HashMap<>();
        
        for (String column : dataset.getColumns()) {
            DataSensitivity maxSensitivity = DataSensitivity.PUBLIC;
            
            // 抽样检测数据敏感度
            for (String value : dataset.sample(column, 100)) {
                DataSensitivity sensitivity = classifyData(value, column);
                if (sensitivity.ordinal() > maxSensitivity.ordinal()) {
                    maxSensitivity = sensitivity;
                }
            }
            
            result.put(column, maxSensitivity);
        }
        
        return result;
    }
}