Spring Batch এবং ডেটাবেস Integration

স্প্রিং ব্যাচ (Spring Batch) একটি শক্তিশালী ফ্রেমওয়ার্ক, যা ব্যাচ প্রসেসিং (যেমন ডেটা মাইগ্রেশন, ডেটা ট্রান্সফরমেশন এবং ডেটা লোডিং) করার জন্য ব্যবহৃত হয়। স্প্রিং ব্যাচ ডেটাবেসের সঙ্গে সহজে ইন্টিগ্রেট করা যায়, এবং এটি ডেটাবেস থেকে ডেটা পড়া, প্রসেস করা এবং ডেটাবেসে রাইট করা সহজ করে তোলে।

স্প্রিং ব্যাচে ডেটাবেস ইন্টিগ্রেশন করার জন্য বিভিন্ন কম্পোনেন্ট যেমন ItemReader, ItemProcessor, ItemWriter, এবং Step ব্যবহৃত হয়। এখানে আমরা একটি উদাহরণ দিয়ে দেখব কিভাবে স্প্রিং ব্যাচের মাধ্যমে ডেটাবেসের সঙ্গে ইন্টিগ্রেশন করা যায়।

স্প্রিং ব্যাচ এবং ডেটাবেস ইন্টিগ্রেশন কনফিগারেশন

ধরা যাক, আমাদের একটি User টেবিল আছে যেখানে id, name, এবং email ফিল্ড রয়েছে, এবং আমরা এই টেবিল থেকে ডেটা রিড, প্রসেস এবং আবার ডেটাবেসে ইনসার্ট করতে চাই।

Step 1: স্প্রিং ব্যাচ কনফিগারেশন

প্রথমে, আমাদের স্প্রিং ব্যাচ কনফিগারেশন তৈরি করতে হবে, যেখানে আমরা Job, Step, এবং ডেটাবেসে ডেটা রিড এবং রাইট করার জন্য ItemReader, ItemProcessor, এবং ItemWriter কনফিগার করব।

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private DataSource dataSource;

    // ItemReader for reading data from the database
    @Bean
    public ItemReader<User> reader() {
        JdbcCursorItemReader<User> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);
        reader.setSql("SELECT id, name, email FROM users WHERE processed = false");
        reader.setRowMapper(new BeanPropertyRowMapper<>(User.class));
        return reader;
    }

    // ItemProcessor for processing data (For example, changing name to uppercase)
    @Bean
    public ItemProcessor<User, User> processor() {
        return new UserItemProcessor();
    }

    // ItemWriter for writing processed data to the database
    @Bean
    public ItemWriter<User> writer() {
        JdbcBatchItemWriter<User> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        writer.setSql("UPDATE users SET name = :name, email = :email, processed = true WHERE id = :id");
        writer.setDataSource(dataSource);
        return writer;
    }

    // Step for the batch job
    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<User, User> chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    // Job for executing the batch process
    @Bean
    public Job importUserJob() {
        return jobBuilderFactory.get("importUserJob")
                .start(step1())
                .build();
    }
}

Step 2: ItemProcessor কাস্টম ক্লাস তৈরি করা

এখানে আমরা একটি ItemProcessor তৈরি করেছি যা প্রতিটি User অবজেক্টের নাম uppercase করে।

public class UserItemProcessor implements ItemProcessor<User, User> {

    @Override
    public User process(User user) throws Exception {
        // Name processing: Change the user's name to uppercase
        user.setName(user.getName().toUpperCase());
        return user;
    }
}

Step 3: User Entity ক্লাস তৈরি করা

এখন, আমাদের একটি User ক্লাস তৈরি করতে হবে, যেটি ডেটাবেসের টেবিলের সঙ্গে মানানসই হবে।

@Entity
public class User {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String name;

    private String email;

    private boolean processed;

    // getters and setters
}

Step 4: ডেটাবেসে users টেবিলের স্কিমা

এখানে একটি সহজ users টেবিলের স্কিমা দেওয়া হলো:

CREATE TABLE users (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    processed BOOLEAN DEFAULT FALSE
);

এখানে processed কলামটি একটি Boolean ভ্যালু ধারণ করে, যা নির্দেশ করে যে ডেটা প্রসেস করা হয়েছে কিনা। প্রথমবার রান করার সময় এটি FALSE থাকবে, এবং প্রসেস হওয়ার পর এটি TRUE হয়ে যাবে।

স্প্রিং ব্যাচ এবং ডেটাবেস ইন্টিগ্রেশনের কাজের প্রক্রিয়া

1. ItemReader:
   • JdbcCursorItemReader ব্যবহার করা হচ্ছে ডেটাবেস থেকে ডেটা রিড করার জন্য। এটি ডেটাবেস থেকে users টেবিলের id, name, এবং email কলাম রিড করবে, যেখানে processed = false।
2. ItemProcessor:
   • UserItemProcessor ক্লাসে, প্রতিটি User অবজেক্টের নামকে uppercase করা হচ্ছে। এটি ডেটাকে প্রসেস করার জন্য ব্যবহৃত হয়।
3. ItemWriter:
   • JdbcBatchItemWriter ব্যবহার করা হচ্ছে ডেটাবেসে ডেটা আপডেট করার জন্য। এটি UPDATE SQL কুয়েরি ব্যবহার করে প্রসেস করা ডেটা ডেটাবেসে সেভ করবে।
4. Chunk Size:
   • chunk(10) এর মাধ্যমে প্রতিটি chunk-এ ১০টি রেকর্ড প্রক্রিয়া করা হবে এবং পরবর্তীতে ডেটাবেসে লেখা হবে।
5. Job:
   • একটি Job তৈরি করা হয়েছে, যা এক বা একাধিক Step নিয়ে গঠিত। এখানে step1 একটি স্টেপ হিসেবে যোগ করা হয়েছে, যা ডেটা রিড, প্রসেস এবং রাইট করার কাজ করবে।

Step 5: ব্যাচ জব রান করা

স্প্রিং ব্যাচ জব রান করতে, JobLauncher ব্যবহার করা হয়। নিচের কোডে দেখানো হয়েছে কিভাবে এটি করা যাবে:

@Autowired
private JobLauncher jobLauncher;

@Autowired
private Job importUserJob;

public void runJob() throws Exception {
    jobLauncher.run(importUserJob, new JobParameters());
}

এখানে jobLauncher.run() মেথডের মাধ্যমে importUserJob রান করা হচ্ছে।

উপসংহার

স্প্রিং ব্যাচ এবং ডেটাবেস ইন্টিগ্রেশন একটি শক্তিশালী পদ্ধতি, যা ডেটা মাইগ্রেশন, ডেটা ট্রান্সফরমেশন এবং ডেটা লোডিং-এর মতো কাজগুলোকে কার্যকরভাবে সম্পন্ন করতে সহায়তা করে। ItemReader, ItemProcessor, এবং ItemWriter কম্পোনেন্ট ব্যবহার করে আপনি ডেটাবেস থেকে ডেটা পড়তে, প্রসেস করতে এবং ডেটাবেসে আপডেট করতে পারেন। এই প্রক্রিয়াটি ডেটার ব্যাচ প্রক্রিয়াকরণের জন্য অত্যন্ত কার্যকর, বিশেষ করে যখন আপনি বড় পরিমাণ ডেটার সঙ্গে কাজ করছেন।

Spring Batch ফ্রেমওয়ার্কে, ডেটাবেস ইন্টিগ্রেশন একটি অত্যন্ত গুরুত্বপূর্ণ অংশ, যা বৃহৎ পরিমাণ ডেটা প্রসেসিং, রিডিং, প্রসেসিং এবং রাইটিংয়ের জন্য ব্যবহৃত হয়। ডেটাবেস ইন্টিগ্রেশন স্প্রিং ব্যাচে একটি Job তৈরি করার প্রক্রিয়ায় সাহায্য করে, যেখানে ডেটাবেসের সাথে মিথস্ক্রিয়া করে ডেটা রিড, প্রসেস এবং আউটপুট ডেটাবেসে বা অন্য কোনো উৎসে সেভ করা হয়। এটি মূলত Data-centric Batch Processing প্রক্রিয়াগুলির জন্য প্রয়োজনীয়, যেখানে ডেটাবেস থেকে একাধিক ধাপে ডেটা পরিচালনা করা হয়।

Spring Batch এ ডেটাবেস ইন্টিগ্রেশন করা হয় ItemReader, ItemProcessor, ItemWriter এবং JdbcCursorItemReader, JdbcBatchItemWriter এর মাধ্যমে, যা ডেটাবেসের সাথে এক্সটেনসিভ ইন্টারঅ্যাকশন প্রদান করে। নিচে আলোচনা করা হবে কেন এবং কিভাবে Spring Batch এ ডেটাবেস ইন্টিগ্রেশন প্রয়োজন এবং কীভাবে এটি কার্যকরভাবে ব্যবহার করা যায়।

Spring Batch এ ডেটাবেস ইন্টিগ্রেশনের প্রয়োজনীয়তা

1. ব্যাচ ডেটা প্রসেসিংয়ের জন্য ডেটাবেসে ডেটার সরাসরি সংযোগ: Spring Batch এ ডেটাবেস ইন্টিগ্রেশন খুবই প্রয়োজনীয়, কারণ অনেক সময়ে ডেটা ডেটাবেসে সংরক্ষিত থাকে এবং সেই ডেটা ব্যাচ প্রসেসিংয়ের জন্য রিড, প্রসেস বা রাইট করতে হয়। স্প্রিং ব্যাচের ItemReader এবং ItemWriter ডেটাবেসে ডেটা রিড এবং রাইট করার জন্য ব্যবহৃত হয়।
2. ডেটাবেস থেকে ডেটা রিড এবং ফিল্টারিং: অনেক ক্ষেত্রেই Batch Processing এর জন্য ডেটাবেস থেকে বিশাল পরিমাণ ডেটা রিড করা হয়। যেমন, ডেটাবেসের টেবিল থেকে ইন্টারেকটিভভাবে ডেটা প্রক্রিয়া করে ইউজারদের জন্য রিপোর্ট তৈরি করা বা বড় আকারের ডেটা আপডেট করা।
3. ডেটাবেস ট্রানজেকশন ম্যানেজমেন্ট: ব্যাচ প্রসেসিংয়ের জন্য ডেটাবেস ট্রানজেকশন ম্যানেজমেন্ট অত্যন্ত গুরুত্বপূর্ণ। Spring Batch JDBC ট্রানজেকশন ম্যানেজমেন্ট সহ ডেটাবেসের সাথে কাজ করতে পারে এবং ব্যাচের পুরো প্রক্রিয়া ট্রানজেকশনালভাবে পরিচালনা করতে সক্ষম, যা ডেটা সঠিকভাবে রিটেন করার জন্য খুবই কার্যকর।
4. ডেটাবেসে ব্যাচের মেটাডেটা সংরক্ষণ: Spring Batch ডেটাবেস ইন্টিগ্রেশন ব্যবহার করলে আমরা JobRepository এর মাধ্যমে ব্যাচ প্রসেসিংয়ের মেটাডেটা সংরক্ষণ করতে পারি। এতে ব্যাচের স্ট্যাটাস, মেটাডেটা এবং জবের প্রগ্রেস ট্র্যাক করা সম্ভব হয়।
5. ডেটা ম্যানিপুলেশন এবং ফিল্টারিং: ব্যাচ প্রসেসিংয়ে JPA বা JDBC ব্যবহার করে ডেটার ম্যানিপুলেশন এবং ফিল্টারিং করা যায়। ডেটাবেসে সংরক্ষিত ডেটার ওপর ItemProcessor ব্যবহার করে কাস্টম ফিল্টার, টান্সফর্মেশন বা পরিবর্তন করা হয় এবং পরবর্তী স্টেপে তা ব্যবহার করা হয়।
6. স্কেলেবিলিটি এবং পারফরম্যান্স: ডেটাবেস ইন্টিগ্রেশন ব্যবহার করে ডেটা Chunk-Oriented Processing এ কার্যকরভাবে রিড, প্রসেস এবং রাইট করা সম্ভব হয়, যা উচ্চ পারফরম্যান্স এবং স্কেলেবিলিটি নিশ্চিত করে।

Spring Batch এ ডেটাবেস ইন্টিগ্রেশন কিভাবে কাজ করে?

Spring Batch এ ডেটাবেস ইন্টিগ্রেশন বাস্তবায়িত করতে ItemReader, ItemProcessor, এবং ItemWriter এর সাহায্যে ডেটাবেসের সাথে যোগাযোগ করা হয়। বিশেষত, ItemReader ডেটা রিড, ItemProcessor ডেটা প্রক্রিয়া এবং ItemWriter ডেটা আউটপুটের জন্য ব্যবহৃত হয়।

১. ItemReader (ডেটা রিড)

Spring Batch ডেটাবেস থেকে ডেটা রিড করার জন্য বিভিন্ন ধরনের ItemReader প্রদান করে, যেমন JdbcCursorItemReader, JpaPagingItemReader, এবং JdbcPagingItemReader।

JdbcCursorItemReader উদাহরণ:

@Bean
public JdbcCursorItemReader<User> userItemReader(DataSource dataSource) {
    JdbcCursorItemReader<User> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(dataSource);
    reader.setSql("SELECT id, first_name, last_name FROM users");
    reader.setRowMapper(new UserRowMapper());
    return reader;
}

২. ItemProcessor (ডেটা প্রক্রিয়া)

ডেটা রিড করার পর, ItemProcessor ব্যবহার করে ডেটা প্রক্রিয়া করা হয়, যেমন ফিল্টার, ট্রান্সফর্ম বা অন্য কোনো পরিবর্তন করা।

@Bean
public ItemProcessor<User, User> userItemProcessor() {
    return user -> {
        // Example transformation: Convert first name to uppercase
        user.setFirstName(user.getFirstName().toUpperCase());
        return user;
    };
}

৩. ItemWriter (ডেটা রাইট)

ItemWriter ব্যবহার করে প্রক্রিয়া করা ডেটা ডেটাবেসে লেখা হয়। যেমন JdbcBatchItemWriter বা JpaItemWriter ব্যবহার করা যেতে পারে।

JdbcBatchItemWriter উদাহরণ:

@Bean
public ItemWriter<User> userItemWriter(DataSource dataSource) {
    JdbcBatchItemWriter<User> writer = new JdbcBatchItemWriter<>();
    writer.setDataSource(dataSource);
    writer.setSql("INSERT INTO users (first_name, last_name) VALUES (:firstName, :lastName)");
    writer.setItemPreparedStatementSetter(new UserPreparedStatementSetter());
    return writer;
}

Spring Batch এ ডেটাবেস ট্রানজেকশন ম্যানেজমেন্ট

Spring Batch ব্যাচ প্রসেসিংয়ের জন্য Transaction Management ব্যবহার করে, যেখানে একটি ব্যাচের সমস্ত স্টেপ একটি একক ট্রানজেকশনে সম্পাদিত হয়। @Transactional অ্যানোটেশন ব্যবহার করে টার্ন-অফ পয়েন্ট তৈরি করা হয়, যাতে ব্যাচের সকল অপারেশন সঠিকভাবে সম্পাদিত হয় এবং কোন ত্রুটি ঘটলে তা পুনরায় চালানো যেতে পারে।

@Transactional
public void executeJob() {
    // Job Execution logic
}

Spring Batch ডেটাবেসে ট্রানজেকশন পরিচালনা করার জন্য TransactionManager এবং JobRepository ব্যবহার করে নিশ্চিত করে যে, ব্যাচের সমস্ত কাজ সঠিকভাবে এবং ডেটার সামঞ্জস্য রক্ষা করে পরিচালিত হচ্ছে।

Spring Batch এবং ডেটাবেস ইন্টিগ্রেশন এর সুবিধা

1. স্কেলেবিলিটি: Spring Batch ব্যাচ প্রসেসিংয়ের জন্য ডেটাবেসের সাথে ইন্টিগ্রেশন করার মাধ্যমে বৃহৎ পরিমাণ ডেটা প্রসেস করা সহজ এবং দ্রুত হয়।
2. ডেটাবেস ট্রানজেকশন ম্যানেজমেন্ট: Spring Batch ডেটাবেসে ব্যাচ অপারেশনগুলো ট্রানজেকশনালভাবে পরিচালনা করে, যাতে কোনো ভুল হলে পুরো প্রসেসটি রোলব্যাক করা যায়।
3. পুনরায় চালানো (Restartability): Spring Batch ডেটাবেস ইন্টিগ্রেশন ব্যবহার করে ব্যাচের কাজ পুনরায় চালানোর সুবিধা দেয়, যা ডেটা সঠিকভাবে প্রক্রিয়াকরণের জন্য অত্যন্ত গুরুত্বপূর্ণ।
4. ডেটা ম্যানিপুলেশন: ডেটাবেস থেকে রিড এবং প্রসেস করার সময় ডেটা ট্রান্সফর্মেশন বা কাস্টম ফিল্টারিং সহজে করা যায়।

সারসংক্ষেপ

Spring Batch এর মাধ্যমে ডেটাবেস ইন্টিগ্রেশন ব্যাচ প্রসেসিংয়ের একটি গুরুত্বপূর্ণ অংশ, যা ডেটাবেস থেকে ডেটা রিড, প্রসেস, এবং রাইট করার জন্য ব্যবহৃত হয়। ItemReader, ItemProcessor, এবং ItemWriter এর মাধ্যমে সহজেই ডেটাবেসের সাথে কাজ করা যায় এবং ডেটা প্রক্রিয়া করা যায়। ডেটাবেস ইন্টিগ্রেশন ব্যবহার করে, Spring Batch ডেটা প্রসেসিংয়ের স্কেলেবিলিটি, ট্রানজেকশন ম্যানেজমেন্ট এবং পুনরায় চালানোর সুবিধা প্রদান করে, যা বৃহৎ পরিমাণ ডেটা পরিচালনার জন্য অত্যন্ত কার্যকর।

Spring Batch ব্যবহার করে ডেটাবেস থেকে ডেটা পড়া এবং লেখা একটি সাধারণ কাজ যা ItemReader এবং ItemWriter এর মাধ্যমে করা হয়। Spring Batch বিশেষভাবে JDBC এবং JPA এর মাধ্যমে ডেটাবেস অপারেশন করতে সহায়তা করে, যা বড় আকারের ডেটা প্রক্রিয়াকরণের জন্য খুবই কার্যকরী।

Spring Batch-এ ডেটাবেস থেকে ডেটা পড়তে এবং ডেটা লিখতে দুটি প্রধান উপাদান থাকে:

• ItemReader: এটি ডেটা সোর্স থেকে ডেটা রিড করে।
• ItemWriter: এটি প্রসেস করা ডেটা ডেটাবেসে বা অন্য কোনো আউটপুট সোর্সে লেখে।

এখানে আমরা JDBC এবং JPA এর মাধ্যমে ডেটাবেস থেকে ডেটা পড়া এবং লেখা সম্পর্কে বিস্তারিত আলোচনা করব।

১. JDBC এর মাধ্যমে ডেটাবেস থেকে ডেটা পড়া এবং লেখা

JDBC (Java Database Connectivity) হলো Java অ্যাপ্লিকেশনের জন্য ডেটাবেসের সাথে যোগাযোগ করার একটি স্ট্যান্ডার্ড পদ্ধতি। Spring Batch-এ JdbcCursorItemReader এবং JdbcBatchItemWriter ব্যবহার করে ডেটাবেস থেকে ডেটা পড়া এবং লেখা হয়।

১.১ JdbcCursorItemReader ব্যবহার করে ডেটা পড়া

JdbcCursorItemReader JDBC কনেকশনের মাধ্যমে ডেটাবেস থেকে ডেটা রিড করতে ব্যবহৃত হয়। এটি SQL কোয়েরি চালিয়ে ডেটাবেস থেকে রেকর্ড রিটার্ন করে।

উদাহরণ: JdbcCursorItemReader ব্যবহার করে ডেটা পড়া

import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.jdbc.datasource.DriverManagerDataSource;

@Bean
public JdbcCursorItemReader<User> reader() {
    JdbcCursorItemReader<User> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(dataSource()); // ডেটাবেস কানেকশন
    reader.setSql("SELECT id, name, email FROM users"); // SQL কোয়েরি
    reader.setRowMapper(new BeanPropertyRowMapper<>(User.class)); // রেকর্ড ম্যাপিং
    return reader;
}

@Bean
public DriverManagerDataSource dataSource() {
    DriverManagerDataSource dataSource = new DriverManagerDataSource();
    dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
    dataSource.setUrl("jdbc:mysql://localhost:3306/mydb");
    dataSource.setUsername("root");
    dataSource.setPassword("password");
    return dataSource;
}

ব্যাখ্যা:

• JdbcCursorItemReader: ডেটাবেস থেকে ডেটা রিড করার জন্য ব্যবহার করা হয়।
• setSql: SQL কোয়েরি সেট করা হয়েছে, যা users টেবিল থেকে ডেটা রিটার্ন করবে।
• setRowMapper: BeanPropertyRowMapper ব্যবহার করে ডেটাবেস রেকর্ডগুলি User অবজেক্টে ম্যাপ করা হয়।

১.২ JdbcBatchItemWriter ব্যবহার করে ডেটা লেখা

JdbcBatchItemWriter ব্যবহৃত হয় ব্যাচ আকারে ডেটা ডেটাবেসে লিখতে।

উদাহরণ: JdbcBatchItemWriter ব্যবহার করে ডেটা লেখা

import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.jdbc.core.BeanPropertyItemPreparedStatementSetter;
import org.springframework.jdbc.core.PreparedStatementSetter;

@Bean
public JdbcBatchItemWriter<User> writer() {
    JdbcBatchItemWriter<User> writer = new JdbcBatchItemWriter<>();
    writer.setDataSource(dataSource()); // ডেটাবেস কানেকশন
    writer.setSql("INSERT INTO users (id, name, email) VALUES (?, ?, ?)"); // SQL কোয়েরি
    writer.setItemPreparedStatementSetter(new BeanPropertyItemPreparedStatementSetter<>()); // প্যারামিটার সেট
    return writer;
}

ব্যাখ্যা:

• JdbcBatchItemWriter: এটি ডেটাবেসে ব্যাচ আকারে ডেটা লিখতে ব্যবহৃত হয়।
• setSql: ইনসার্ট SQL কোয়েরি দিয়ে ডেটাবেসে ডেটা লেখা হয়।
• setItemPreparedStatementSetter: এটি User অবজেক্টের প্রপার্টি মেপিং করে SQL প্যারামিটার পাঠানোর জন্য ব্যবহৃত হয়।

২. JPA এর মাধ্যমে ডেটাবেস থেকে ডেটা পড়া এবং লেখা

JPA (Java Persistence API) একটি Java API যা ডেটাবেসের সাথে অবজেক্ট-রিলেশনাল ম্যাপিং (ORM) করে। Spring Batch-এ JpaPagingItemReader এবং JpaItemWriter ব্যবহৃত হয় ডেটাবেস থেকে ডেটা রিড এবং লেখার জন্য।

২.১ JpaPagingItemReader ব্যবহার করে ডেটা পড়া

JpaPagingItemReader JPA (Hibernate) ব্যবহার করে পেজ বাই পেজ ডেটা রিড করতে ব্যবহৃত হয়। এটি JPA কনফিগারেশন এবং @Entity ক্লাসের মাধ্যমে ডেটাবেস থেকে ডেটা রিড করে।

উদাহরণ: JpaPagingItemReader ব্যবহার করে ডেটা পড়া

import org.springframework.batch.item.database.JpaPagingItemReader;
import org.springframework.batch.item.database.JpaQueryProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.orm.jpa.JpaTransactionManager;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;

@Bean
public JpaPagingItemReader<User> reader() {
    JpaPagingItemReader<User> reader = new JpaPagingItemReader<>();
    reader.setEntityManagerFactory(entityManagerFactory().getObject());
    reader.setQueryProvider(queryProvider()); // JPA কোয়েরি
    reader.setPageSize(10); // পেজ সাইজ
    return reader;
}

@Bean
public LocalContainerEntityManagerFactoryBean entityManagerFactory() {
    // আপনার JPA EntityManagerFactory কনফিগারেশন
}

@Bean
public JpaQueryProvider queryProvider() {
    // JPQL কোয়েরি তৈরি করুন
}

ব্যাখ্যা:

• JpaPagingItemReader: এটি JPA ব্যবহার করে ডেটাবেস থেকে পেজ-বাই-পেজ ডেটা রিড করতে ব্যবহৃত হয়।
• setEntityManagerFactory: EntityManagerFactory কনফিগারেশন প্রদান করা হয়।
• setQueryProvider: JPQL কোয়েরি প্রদান করা হয়, যা ডেটাবেস থেকে রেকর্ড ফেচ করবে।

২.২ JpaItemWriter ব্যবহার করে ডেটা লেখা

JpaItemWriter ব্যবহৃত হয় প্রসেস করা ডেটা ডেটাবেসে সেভ করতে। এটি JPA EntityManager ব্যবহার করে ডেটা সেভ করে।

উদাহরণ: JpaItemWriter ব্যবহার করে ডেটা লেখা

import org.springframework.batch.item.database.JpaItemWriter;
import org.springframework.context.annotation.Bean;

@Bean
public JpaItemWriter<User> writer() {
    JpaItemWriter<User> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(entityManagerFactory().getObject());
    return writer;
}

ব্যাখ্যা:

• JpaItemWriter: এটি JPA EntityManager ব্যবহার করে ডেটাবেসে ডেটা সেভ করতে ব্যবহৃত হয়।
• setEntityManagerFactory: EntityManagerFactory সেট করে যা ডেটাবেসে ডেটা সেভ করতে ব্যবহৃত হয়।

সারাংশ

JDBC এবং JPA এর মাধ্যমে Spring Batch ডেটাবেস থেকে ডেটা পড়া এবং লেখা অত্যন্ত কার্যকরী ও সুবিধাজনক। JDBC ব্যবহার করে সরাসরি SQL কোয়েরি দিয়ে ডেটা রিড এবং রাইট করা হয়, যেখানে JPA ORM (Object-Relational Mapping) ব্যবহার করে ডেটা অবজেক্টের মাধ্যমে ডেটাবেসের সাথে যোগাযোগ স্থাপন করে। JdbcCursorItemReader, JdbcBatchItemWriter, JpaPagingItemReader, এবং JpaItemWriter Spring Batch-এ জনপ্রিয় উপাদান যা ডেটাবেসে ডেটা প্রক্রিয়া (Read and Write) করতে ব্যবহৃত হয়।

স্প্রিং ব্যাচ (Spring Batch) ডেটাবেস ইন্টিগ্রেশন খুবই গুরুত্বপূর্ণ একটি বৈশিষ্ট্য, যা ব্যাচ প্রসেসিংয়ের সময় ডাটাবেসে ডাটা রিডিং, প্রসেসিং, এবং রাইটিংয়ের কাজ করতে সাহায্য করে। ডেটাবেস ইন্টিগ্রেশন সাধারণত ItemReader, ItemProcessor, এবং ItemWriter এর মাধ্যমে পরিচালিত হয়, যেখানে ItemReader ডাটাবেস থেকে ডাটা পড়তে, ItemProcessor ডাটাকে প্রসেস করতে এবং ItemWriter সেই ডাটা ডাটাবেসে সেভ করতে ব্যবহৃত হয়।

স্প্রিং ব্যাচের ডেটাবেস ইন্টিগ্রেশন সহজ করতে বিভিন্ন ধরনের ItemReader এবং ItemWriter রয়েছে, যেমন JdbcCursorItemReader, JpaPagingItemReader, এবং JdbcBatchItemWriter। এখানে আমরা ডেটাবেসের সাথে কাজ করার জন্য একটি সাধারণ উদাহরণ দেখব।

ডেটাবেস ইন্টিগ্রেশন উদাহরণ

উদাহরণ: Employee ডাটা প্রসেসিং

ধরা যাক, আমাদের একটি Employee টেবিল রয়েছে, যেখানে কর্মচারীদের নাম, ইমেইল, এবং তাদের বেতন রয়েছে। আমাদের উদ্দেশ্য হল, এই কর্মচারীদের বেতনকে কিছু বৃদ্ধি করে আবার ডাটাবেসে সেভ করা।

১. ডেটাবেস টেবিল তৈরি

প্রথমে, একটি Employee টেবিল তৈরি করি, যেটি SQL ডেটাবেসে ব্যবহার করা হবে:

CREATE TABLE employee (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    salary DECIMAL
);

২. স্প্রিং ব্যাচ কনফিগারেশন

স্প্রিং ব্যাচে ডেটাবেস ইন্টিগ্রেশন কনফিগারেশন করার জন্য প্রথমে ItemReader, ItemProcessor, এবং ItemWriter কনফিগার করতে হবে।

২.১. JdbcCursorItemReader - ডেটাবেস থেকে ডাটা পড়া

JdbcCursorItemReader ডেটাবেস থেকে রেকর্ড একে একে রিড করতে ব্যবহৃত হয়। আমরা এই ক্লাসটি ব্যবহার করে কর্মচারীদের ডাটা রিড করব।

import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import javax.sql.DataSource;

@Bean
public JdbcCursorItemReader<Employee> reader(DataSource dataSource) {
    JdbcCursorItemReader<Employee> reader = new JdbcCursorItemReader<>();
    reader.setDataSource(dataSource);
    reader.setSql("SELECT id, name, email, salary FROM employee");
    reader.setRowMapper(new BeanPropertyRowMapper<>(Employee.class));
    return reader;
}

এখানে, JdbcCursorItemReader SQL কুয়েরি ব্যবহার করে Employee টেবিল থেকে সমস্ত কর্মচারী ডাটা রিড করছে। BeanPropertyRowMapper ক্লাসটি ডাটাবেস রেজাল্ট সেটকে Employee অবজেক্টে ম্যাপ করে।

২.২. ItemProcessor - ডাটা প্রসেসিং

এখন, কর্মচারীদের বেতন ১০% বাড়ানোর জন্য একটি ItemProcessor তৈরি করি:

import org.springframework.batch.item.ItemProcessor;

public class EmployeeSalaryProcessor implements ItemProcessor<Employee, Employee> {

    @Override
    public Employee process(Employee item) throws Exception {
        // বেতন ১০% বৃদ্ধি
        item.setSalary(item.getSalary() * 1.1);
        return item;
    }
}

এই ItemProcessor ক্লাসটি প্রতিটি Employee অবজেক্টের বেতন ১০% বৃদ্ধি করবে।

২.৩. JdbcBatchItemWriter - ডেটাবেসে ডাটা লেখানো

এখন, JdbcBatchItemWriter ব্যবহার করে প্রসেস করা ডাটা ডাটাবেসে সেভ করা হবে।

import org.springframework.batch.item.database.JdbcBatchItemWriter;
import javax.sql.DataSource;

@Bean
public JdbcBatchItemWriter<Employee> writer(DataSource dataSource) {
    JdbcBatchItemWriter<Employee> writer = new JdbcBatchItemWriter<>();
    writer.setDataSource(dataSource);
    writer.setSql("UPDATE employee SET name = ?, email = ?, salary = ? WHERE id = ?");
    writer.setItemPreparedStatementSetter(new EmployeePreparedStatementSetter());
    return writer;
}

এখানে, JdbcBatchItemWriter SQL UPDATE কুয়েরি ব্যবহার করে ডাটাবেসের Employee টেবিল আপডেট করবে। EmployeePreparedStatementSetter ক্লাসটি ডাটা সঠিকভাবে PreparedStatement এ সেট করবে।

EmployeePreparedStatementSetter ক্লাসের উদাহরণ:

import java.sql.PreparedStatement;
import java.sql.SQLException;
import org.springframework.batch.item.database.ItemPreparedStatementSetter;

public class EmployeePreparedStatementSetter implements ItemPreparedStatementSetter<Employee> {

    @Override
    public void setValues(Employee item, PreparedStatement ps) throws SQLException {
        ps.setString(1, item.getName());
        ps.setString(2, item.getEmail());
        ps.setBigDecimal(3, item.getSalary());
        ps.setInt(4, item.getId());
    }
}

এটি প্রতিটি Employee অবজেক্টের ডাটা সঠিকভাবে PreparedStatement এর মাধ্যমে ডাটাবেসে আপডেট করবে।

৩. Job এবং Step কনফিগারেশন

এখন, সবগুলো উপাদানকে একত্রিত করে একটি Job এবং Step কনফিগার করা হবে:

import org.springframework.batch.core.Step;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class BatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Job updateEmployeeSalaryJob() {
        return jobBuilderFactory.get("updateEmployeeSalaryJob")
                .start(processEmployeeSalaryStep())
                .build();
    }

    @Bean
    public Step processEmployeeSalaryStep() {
        return stepBuilderFactory.get("processEmployeeSalaryStep")
                .<Employee, Employee>chunk(10)
                .reader(reader(null))  // Reader: JdbcCursorItemReader
                .processor(new EmployeeSalaryProcessor())  // Processor: EmployeeSalaryProcessor
                .writer(writer(null))  // Writer: JdbcBatchItemWriter
                .build();
    }
}

এখানে, updateEmployeeSalaryJob জবটি একটি স্টেপ processEmployeeSalaryStep নিয়ে গঠিত। এই স্টেপে JdbcCursorItemReader, EmployeeSalaryProcessor, এবং JdbcBatchItemWriter যুক্ত রয়েছে।

সারাংশ

স্প্রিং ব্যাচে ডেটাবেস ইন্টিগ্রেশন একটি গুরুত্বপূর্ণ প্রক্রিয়া, যেখানে ItemReader, ItemProcessor, এবং ItemWriter ব্যবহার করে ডাটাবেস থেকে ডাটা রিড, প্রসেস এবং রাইট করা হয়। এই উদাহরণে JdbcCursorItemReader, ItemProcessor, এবং JdbcBatchItemWriter ব্যবহার করা হয়েছে কর্মচারীদের বেতন ১০% বৃদ্ধি করার জন্য এবং সেই ডাটা ডাটাবেসে আপডেট করা হয়েছে। স্প্রিং ব্যাচের মাধ্যমে ডেটাবেস ইন্টিগ্রেশন সহজ এবং কার্যকরীভাবে করা যায়, যা বড় পরিমাণ ডাটার প্রসেসিংয়ের জন্য অত্যন্ত উপযোগী।