Что такое ETL?

ETL: Extract, Transform, Load

Что это такое

ETL (Extract, Transform, Load) — это процесс обработки данных, состоящий из трёх этапов:

1. Extract (Извлечение) — получение данных из исходных источников
2. Transform (Трансформация) — обработка, очистка и преобразование данных
3. Load (Загрузка) — сохранение обработанных данных в целевую систему

Это одна из ключевых концепций в Data Engineering и Data Warehouse архитектуре.

Компоненты ETL процесса

1. Extract (Извлечение)

Получение данных из различных источников:

public class DataExtractor {
    
    // Извлечение из базы данных
    public List<Customer> extractFromDatabase() throws SQLException {
        List<Customer> customers = new ArrayList<>();
        try (Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/shop")) {
            String query = "SELECT * FROM customers";
            ResultSet rs = conn.createStatement().executeQuery(query);
            while (rs.next()) {
                customers.add(new Customer(
                    rs.getInt("id"),
                    rs.getString("name"),
                    rs.getString("email")
                ));
            }
        }
        return customers;
    }
    
    // Извлечение из API
    public List<Order> extractFromAPI() throws IOException {
        String apiUrl = "https://api.example.com/orders";
        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(apiUrl))
            .header("Authorization", "Bearer token")
            .GET()
            .build();
        
        HttpResponse<String> response = client.send(request, 
            HttpResponse.BodyHandlers.ofString());
        
        return parseJson(response.body());
    }
    
    // Извлечение из файла
    public List<Product> extractFromCSV(String filePath) throws IOException {
        List<Product> products = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(
            new FileReader(filePath))) {
            
            String line;
            while ((line = reader.readLine()) != null) {
                String[] parts = line.split(",");
                products.add(new Product(
                    Integer.parseInt(parts[0]),
                    parts[1],
                    Double.parseDouble(parts[2])
                ));
            }
        }
        return products;
    }
}

2. Transform (Трансформация)

Обработка и очистка данных:

public class DataTransformer {
    
    public List<CustomerRecord> transformCustomers(List<Customer> rawData) {
        return rawData.stream()
            .filter(c -> c.isActive())
            .map(c -> new Customer(
                c.getId(),
                c.getName().trim().toUpperCase(),
                c.getEmail().toLowerCase()
            ))
            .filter(c -> isValidEmail(c.getEmail()))
            .map(c -> new CustomerRecord(
                c.getId(),
                c.getName(),
                c.getEmail(),
                calculateLoyaltyScore(c),
                LocalDateTime.now()
            ))
            .collect(Collectors.toList());
    }
    
    private boolean isValidEmail(String email) {
        return email.matches("^[A-Za-z0-9+_.-]+@(.+)$");
    }
    
    private int calculateLoyaltyScore(Customer c) {
        return (int) (c.getOrderCount() * 10 + c.getTotalSpent() / 100);
    }
}

3. Load (Загрузка)

Сохранение обработанных данных:

public class DataLoader {
    
    public void loadToDatabase(List<CustomerRecord> records) throws SQLException {
        try (Connection conn = DriverManager.getConnection(
            "jdbc:mysql://localhost:3306/warehouse")) {
            
            String sql = "INSERT INTO customer_warehouse (id, name, email, loyalty_score, load_date) VALUES (?, ?, ?, ?, ?)";
            
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                for (CustomerRecord record : records) {
                    stmt.setInt(1, record.getId());
                    stmt.setString(2, record.getName());
                    stmt.setString(3, record.getEmail());
                    stmt.setInt(4, record.getLoyaltyScore());
                    stmt.setTimestamp(5, Timestamp.valueOf(record.getLoadDate()));
                    stmt.addBatch();
                }
                stmt.executeBatch();
            }
        }
    }
}

Полный ETL Pipeline

public class ETLPipeline {
    
    private DataExtractor extractor = new DataExtractor();
    private DataTransformer transformer = new DataTransformer();
    private DataLoader loader = new DataLoader();
    
    public void runETL() throws Exception {
        System.out.println("[ETL] Начинаем процесс...");
        
        // EXTRACT
        System.out.println("[EXTRACT] Извлекаем данные...");
        List<Customer> rawCustomers = extractor.extractFromDatabase();
        List<Order> rawOrders = extractor.extractFromAPI();
        
        // TRANSFORM
        System.out.println("[TRANSFORM] Трансформируем данные...");
        List<CustomerRecord> transformedCustomers = 
            transformer.transformCustomers(rawCustomers);
        
        // LOAD
        System.out.println("[LOAD] Загружаем в хранилище...");
        loader.loadToDatabase(transformedCustomers);
        System.out.println("[ETL] Процесс завершён");
    }
}

ETL инструменты

• Apache Spark — распределённая обработка данных
• Apache Airflow — оркестрация пайплайнов
• Talend — коммерческий ETL инструмент
• Kettle (PDI) — открытый ETL инструмент
• Spring Batch — для Java приложений

Преимущества ETL

• Централизованная обработка данных
• Масштабируемость
• Гарантия качества данных
• Автоматизация повторяющихся процессов
• Отделение логики от источников данных

Заключение

ETL — фундаментальный процесс в системах обработки данных. Как Java Developer ты можешь создавать ETL пайплайны используя Spring Batch, Apache Spark или собственное решение на Java.