使用Maven和SpringBoot搭建客户数据清洗项目框架

<!-- pom.xml：添加依赖 -->  
<dependencies>  
  <!-- 数据库连接 -->  
  <dependency>  
    <groupId>com.zaxxer</groupId>  
    <artifactId>HikariCP</artifactId>  
    <version>5.0.1</version>  
  </dependency>  
  <!-- 函数式编程工具 -->  
  <dependency>  
    <groupId>org.apache.commons</groupId>  
    <artifactId>commons-lang3</artifactId>  
    <version>3.12.0</version>  
  </dependency>  
  <!-- 数据验证 -->  
  <dependency>  
    <groupId>org.assertj</groupId>  
    <artifactId>assertj-core</artifactId>  
    <version>3.24.2</version>  
    <scope>test</scope>  
  </dependency>  
</dependencies>  

# application.properties：连接MySQL  
spring.datasource.url=jdbc:mysql://localhost:3306/customer_db  
spring.datasource.username=root  
spring.datasource.password=root  
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver  

// CustomerDAO.java：从MySQL读取客户数据  
import java.sql.*;  
import java.util.*;  

public class CustomerDAO {  
    public List<Map<String, Object>> fetchDirtyData() {  
        String query = "SELECT * FROM customers WHERE status = 'pending'";  
        List<Map<String, Object>> data = new ArrayList<>();  

        try (Connection conn = DriverManager.getConnection("jdbc:mysql://...", "...", "...");  
             Statement stmt = conn.createStatement();  
             ResultSet rs = stmt.executeQuery(query)) {  

            ResultSetMetaData meta = rs.getMetaData();  
            while (rs.next()) {  
                Map<String, Object> row = new HashMap<>();  
                for (int i = 1; i <= meta.getColumnCount(); i++) {  
                    row.put(meta.getColumnName(i), rs.getObject(i));  
                }  
                data.add(row);  
            }  
        } catch (SQLException e) {  
            e.printStackTrace();  
        }  
        return data;  
    }  
}  

// ApiService.java：调用第三方API获取数据  
import java.net.http.*;  
import java.net.URI;  
import java.util.concurrent.*;  

public class ApiService {  
    public String fetchDataFromAPI() {  
        HttpClient client = HttpClient.newHttpClient();  
        HttpRequest request = HttpRequest.newBuilder()  
            .uri(URI.create("https://api.example.com/customers"))  
            .build();  

        try {  
            return client.send(request, BodyHandlers.ofString()).body();  
        } catch (Exception e) {  
            return "{}"; // 返回空对象避免崩溃  
        }  
    }  
}  

// DataCleaner.java：清洗单个字段  
import org.apache.commons.lang3.StringUtils;  

public class DataCleaner {  
    public static String cleanPhone(String rawPhone) {  
        // 去除非数字字符  
        String cleaned = StringUtils.remove(rawPhone, " ").replaceAll("[^0-9]", "");  
        // 补全11位手机号  
        if (cleaned.length() == 10) {  
            return "1" + cleaned;  // 假设国内手机号  
        }  
        return cleaned;  
    }  
}  

// BatchProcessor.java：并行处理数据  
public class BatchProcessor {  
    public static List<Map<String, Object>> cleanData(List<Map<String, Object>> rawData) {  
        return rawData.parallelStream()  
            .filter(row -> row.get("email") != null)  // 过滤缺失邮箱  
            .map(row -> {  
                Map<String, Object> cleanedRow = new HashMap<>(row);  
                cleanedRow.put("phone", DataCleaner.cleanPhone((String) row.get("phone")));  
                cleanedRow.put("address", row.get("address").toString().trim());  // 去除前后空格  
                return cleanedRow;  
            })  
            .distinct()  // 去重（需重写equals/hashCode）  
            .collect(Collectors.toList());  
    }  
}  

// DataTransformer.java：使用Jackson库  
import com.fasterxml.jackson.databind.ObjectMapper;  

public class DataTransformer {  
    public String toJson(List<Map<String, Object>> data) {  
        ObjectMapper mapper = new ObjectMapper();  
        try {  
            return mapper.writeValueAsString(data);  
        } catch (Exception e) {  
            return "[]";  
        }  
    }  
}  

// HdfsWriter.java：用Hadoop API  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.*;  

public class HdfsWriter {  
    public void writeToFile(String data) {  
        Configuration conf = new Configuration();  
        Path path = new Path("/user/hadoop/cleaned_customers.json");  

        try (FSDataOutputStream out = FileSystem.get(conf).create(path)) {  
            out.write(data.getBytes());  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }  
}  

// DataValidator.java：校验手机号格式  
public class DataValidator {  
    public static boolean validatePhone(String phone) {  
        // 中国手机号正则：1开头，11位数字  
        return phone != null && phone.matches("^1\\d{10}$");  
    }  
}  

// QualityReport.java：统计清洗结果  
public class QualityReport {  
    public static void generateReport(List<Map<String, Object>> data) {  
        long validPhones = data.stream()  
            .filter(row -> DataValidator.validatePhone((String) row.get("phone")))  
            .count();  

        System.out.println("✅ 有效手机号数量：" + validPhones);  
        System.out.println("⚠️ 总数据量：" + data.size());  
        System.out.println("📊 清洗完成！");  
    }  
}  

// Main.java：端到端流程  
public class Main {  
    public static void main(String[] args) {  
        // 1. 采集数据  
        CustomerDAO dao = new CustomerDAO();  
        List<Map<String, Object>> raw = dao.fetchDirtyData();  

        // 2. 清洗数据  
        List<Map<String, Object>> cleaned = BatchProcessor.cleanData(raw);  

        // 3. 验证数据  
        QualityReport.generateReport(cleaned);  

        // 4. 输出到HDFS  
        HdfsWriter writer = new HdfsWriter();  
        writer.writeToFile(new DataTransformer().toJson(cleaned));  

        System.out.println("🎉 数据清洗大业完成！");  
    }  
}  

// 使用分页查询  
public List<Map<String, Object>> fetchDirtyData(int offset, int limit) {  
    String query = "SELECT * FROM customers WHERE status = 'pending' LIMIT " + offset + ", " + limit;  
    // ...  
}  

// 使用Optional处理可能缺失的字段  
public String getSafeField(JsonNode node, String key) {  
    return node.has(key) ? node.get(key).asText() : "";  
}