<job id="phonebilling" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
     version="1.0">
    <properties>
        <property name="log_file_name" value="log1.txt"/>
        <property name="airtime_price" value="0.08"/>
        <property name="tax_rate" value="0.07"/>
    </properties>
    <step id="callrecords" next="bills"> ... </step>
    <step id="bills"> ... </step>
</job>

<step id="callrecords" next="bills">
    <chunk checkpoint-policy="item" item-count="10">
        <reader ref="CallRecordReader"></reader>
        <processor ref="CallRecordProcessor"></processor>
        <writer ref="CallRecordWriter"></writer>
    </chunk>
</step>

<step id="bills">
    <chunk checkpoint-policy="item" item-count="2">
        <reader ref="BillReader">
            <properties>                <property name="firstItem" value="#{partitionPlan['firstItem']}"/>                <property name="numItems" value="#{partitionPlan['numItems']}"/>            </properties>        </reader>
        <processor ref="BillProcessor"></processor>
        <writer ref="BillWriter"></writer>
    </chunk>
    <partition>
        <mapper ref="BillPartitionMapper"/>
    </partition>
    <end on="COMPLETED"/>
</step>

@Entity
public class CallRecord implements Serializable {
    @Id @GeneratedValue
    private Long id;
    @Temporal(TemporalType.DATE)
    private Date datetime;
    private String fromNumber;
    private String toNumber;
    private int minutes;
    private int seconds;
    private BigDecimal price;

    public CallRecord() { }

    public CallRecord(String datetime, String from,
            String to, int min, int sec)             throws ParseException { ... }

    public CallRecord(String jsonData) throws ParseException { ... }

    /* ... Геттеры и сеттеры ... */
}

{"datetime":"03/01/2013 04:03","from":"555-0101",
"to":"555-0114","length":"03:39"}

@Entity
public class PhoneBill implements Serializable {
    @Id
    private String phoneNumber;
    @OneToMany(fetch = FetchType.EAGER, cascade = CascadeType.PERSIST)
    @OrderBy("datetime ASC")
    private List<CallRecord> calls;
    private BigDecimal amountBase;
    private BigDecimal taxRate;
    private BigDecimal tax;
    private BigDecimal amountTotal;

    public PhoneBill() { }

    public PhoneBill(String number) {
        this.phoneNumber = number;
        calls = new ArrayList<>();
    }

    public void addCall(CallRecord call) {
        calls.add(call);
    }

    public void calculate(BigDecimal taxRate) { ... }

    /* ... Геттеры и сеттеры ... *
}

@Dependent
@Named("CallRecordReader")
public class CallRecordReader implements ItemReader {
    private ItemNumberCheckpoint checkpoint;
    private String fileName;
    private BufferedReader breader;
    @Inject
    JobContext jobCtx;

    /* ... Переопределение методов open, close, readItem
     * и checkpointInfo ... * /
}

fileName = jobCtx.getProperties().getProperty("log_file_name");
breader = new BufferedReader(new FileReader(fileName));

@Override
public Object readItem() throws Exception {
    /* Считывание строки из файла журнала и
     * создание CallRecord из JSON */
    String callEntryJson = breader.readLine();
    if (callEntryJson != null) {
        checkpoint.nextItem();
        return new CallRecord(callEntryJson);
    } else
        return null;
}

@Override
public void writeItems(List<Object> callList) throws Exception {

    for (Object callObject : callList) {
        CallRecord call = (CallRecord) callObject;
        PhoneBill bill = em.find(PhoneBill.class, call.getFromNumber());
        if (bill == null) {
            /* Создание счёта для этого клиента */
            bill = new PhoneBill(call.getFromNumber());
            bill.addCall(call);
            em.persist(bill);
        } else {
            /* Добавление звонка в существующий счёт */
            bill.addCall(call);
        }
    }
}

@Override
public Properties[] getPartitionProperties() {
    /* Назначаем (приблизительно) равное количество элементов
     * в каждое разделение. */
    long totalItems = getBillCount();
    long partItems = (long) totalItems / getPartitions();
    long remItems = totalItems % getPartitions();

    /* Заполняем массив свойств. Каждый элемент Properties
* в массиве соответствует разделению. */
    Properties[] props = new Properties[getPartitions()];

    for (int i = 0; i < getPartitions(); i++) {
        props[i] = new Properties();
        props[i].setProperty("firstItem",
                String.valueOf(i * partItems));
        /* Последнее разделение получает оставшиеся элементы */
        if (i == getPartitions() - 1) {
            props[i].setProperty("numItems",
                    String.valueOf(partItems + remItems));
        } else {
            props[i].setProperty("numItems",
                    String.valueOf(partItems));
    }
    return props;
}

@Dependent
@Named("BillReader")
public class BillReader implements ItemReader {
    @Inject    @BatchProperty(name = "firstItem")    private String firstItemValue;    @Inject    @BatchProperty(name = "numItems")    private String numItemsValue;
    private ItemNumberCheckpoint checkpoint;    @PersistenceContext    private EntityManager em;    private Iterator iterator;

    @Override
    public void open(Serializable ckpt) throws Exception {
        /* Получение диапазона элементов для обработки в этом разделении */
        long firstItem0 = Long.parseLong(firstItemValue);
        long numItems0 = Long.parseLong(numItemsValue);

        if (ckpt == null) {
            /* Создание контрольной точки для этого разделения */
            checkpoint = new ItemNumberCheckpoint();
            checkpoint.setItemNumber(firstItem0);
            checkpoint.setNumItems(numItems0);
        } else {
            checkpoint = (ItemNumberCheckpoint) ckpt;
        }

        /* Adjust range for this partition from the checkpoint */
        long firstItem = checkpoint.getItemNumber();
        long numItems = numItems0 - (firstItem - firstItem0);
        ...
    }
    ...
}

/* Получение итератора для квитанций в этом разделении */
String query = "SELECT b FROM PhoneBill b ORDER BY b.phoneNumber";
Query q = em.createQuery(query).setFirstResult((int) firstItem)
        .setMaxResults((int) numItems);
iterator = q.getResultList().iterator();

<body>
    <h1>The Phone Billing Example Application</h1>
    <h2>Log file</h2>
    <p>The batch job analyzes the following log file:</p>
    <textarea cols="90" rows="25"
              readonly="true">#{jsfBean.createAndShowLog()}</textarea>
    <p> </p>
    <h:form>
        <h:commandButton value="Start Batch Job"
                         action="#{jsfBean.startBatchJob()}" />
    </h:form>
</body>

<p>Current Status of the Job: <b>#{jsfBean.jobStatus}</b></p>
<h:dataTable var="_row" value="#{jsfBean.rowList}"
             border="1" rendered="#{jsfBean.completed}">
    <!-- ... отображение результатов из jsfBean.rowList ... -->
</h:dataTable>
<!-- Render the check status button if the job has not finished -->
<h:form>
    <h:commandButton value="Check Status"
                     rendered="#{jsfBean.completed==false}"
                     action="jobstarted" />
</h:form>

/* Отправка пакетного задания на выполнение.
 * Метод навигации JSF (возвращает название следующей страницы) */
public String startBatchJob() {
    jobOperator = BatchRuntime.getJobOperator();
    execID = jobOperator.start("phonebilling", null);
    return "jobstarted";
}

/ * Получение статуса задания из пакетной среды выполнения * /
public String getJobStatus() {
    return jobOperator.getJobExecution(execID).getBatchStatus().toString();
}