[SPRINGBATCH] Lecture des enregistrements par client depuis un fichier

**ELSEV** · 25/10/2019, 16h09

Bonjour,

Je souhaite lire les enregistrements d'un fichier csv, par Client afin de traiter, par la suite, les informations dans l'ItemProcessor avec l'Objet Client, pour ensuite écrire les données du client en base de données

Versions :
- java 1.8
- springBatch : 3.0.7.RELEASE

Ci-dessous le fichier client CSV
client1;adresse;adresse1
client1;telephone;telephone1;telephone1
client1;facture;facture1
client2;adresse;adresse1;adresse2
client2;telephone;telephone1
client2;facture;facture1
client3;adresse;adresse1
client3;telephone;telephone1
client3;facture;facture1;facture2
EndOfClients

J'utilise RecordSeparatorPolicy pour récupérer les données par client
Avec la méthode isEndOfRecord, j'indique le changement de client pour transmettre les données, via la méthode postProcess, au LineMapper ci dessous les enregistrements passés par la méthode :
postProcess = client1;adresse;adresse1||client1;telephone;telephone1;telephone1||client1;facture;facture1
postProcess = client2;adresse;adresse1;adresse2||client2;telephone;telephone1||client2;facture;facture1
postProcess = client3;adresse;adresse1||client3;telephone;telephone1||client3;facture;facture1;facture2

Avant la concaténation des enregistrements, j'insére un séparateur d'enregistrement "||" afin de pouvoir décomposer
mes enregistrements et construire mon objet Client dans le LineMapper
LineMapper : client1;adresse;adresse1||client1;telephone;telephone1;telephone1||client1;facture;facture1
LineMapper : client2;adresse;adresse1;adresse2||client2;telephone;telephone1||client2;facture;facture1
LineMapper : client3;adresse;adresse1||client3;telephone;telephone1||client3;facture;facture1;facture2

Et écrire dans l'ItemWriter avec l'objet Client
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture

Le soucis est que je suis obligé d'ajouter dans mon fichier csv un indicateur de fin de ligne : EndOfClients
afin que le client3 soit transmis vers le LineMapper, sinon j'obtiens l'erreur suivante

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

org.springframework.batch.item.file.FlatFileParseException: Unexpected end of file before record complete

Tout d'abord, est-ce que j'utilise la bonne manière de faire ?
Si la manière est bonne, comment puis-je détecter que je suis en fin de ligne afin de transmettre le dernier client3 au LineMapper, sans utiliser EndOfClients ?

Ci-dessous le code du ClientRecordSeparatorPolicy

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
public class ClientRecordSeparatorPolicy implements RecordSeparatorPolicy {
 
	// Separateur de champ d'un enregistrement
	private String fieldSeparator = ";";
	// Separateur d'enregistrement
	private String recordSeparator = "||";
	private String END_OF_CLIENT = "EndOfClients";
 
	private DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer(fieldSeparator);
	private String previousClient;
	private List<String> listLineClient = new ArrayList<String>();
 
 
	public boolean isEndOfRecord(String record) {
		FieldSet fieldSet = tokenizer.tokenize(record);
		String currentClient;
 
		// Test si fin de fichier
		if (END_OF_CLIENT.equalsIgnoreCase(record)) {
			return true;
		}	
 
		// Test si init
		currentClient = fieldSet.readString(0);
		if (previousClient != null) {
 
			// Test rupture
			if (!currentClient.equalsIgnoreCase(previousClient)) {
				previousClient = currentClient;
				return true;
			}
 
		} else {
			// Init 1ier record
			previousClient = currentClient;
		}
 
//		System.out.println("isEndOfRecord = " + record);
		return false;
	}
 
	public String postProcess(String record) {
		String clientRecords = listLineClient.stream().collect(Collectors.joining(recordSeparator));
		listLineClient.clear();
		listLineClient.add(record);
		System.out.println("postProcess = " + clientRecords);
		return clientRecords;
	}
 
	public String preProcess(String record) {
		listLineClient.add(record);
//		System.out.println("preProcess = " + record);
		return "";
	}

Sinon dois-je faire autrement ?
Dois-je m'y prendre directement dans l'ItemReader ?
Si oui, auriez vous un exemple ?

Merci

**atha2** · 26/10/2019, 01h27

As-tu essayé d'ajouter un retour à la ligne à la fin de ton fichier ?

Sinon ce que tu fais me parait très compliqué pour lire un simple fichier CSV, normalement tu devrait juste faire :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

csvReader.setDelimiter(";")//ou un truc du genre

Enlève également "EndOfClients" de ton fichier.

**ELSEV** · 28/10/2019, 10h43

Oui il y a bien un retour à la ligne dans le fichier
En retirant la ligne "EndOfClients", il est possible de mettre un blanc à la place comme ci dessous

Nom : 2019-10-28 10_33_26-Window.png
Affichages : 1789
Taille : 5,7 Ko

Nom : 2019-10-28 10_33_26-Window.png
Affichages : 1789
Taille : 5,7 Ko

En contre partie, il me faut détecter la ligne vide comme si dessous

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
 
// Test si fin de fichier
if (END_OF_CLIENT.equalsIgnoreCase(record) || "".equals(record)) {
	return true;
}

Sinon j'ai toujours l'erreur

Je pense qu'il y a bien une solution comme j'ai pu voir dans le doc de Spring
https://docs.spring.io/spring-batch/...l#customReader

Il faut, je pense, customiser la lecture des enregistrements à la manière de csvreader

**atha2** · 30/10/2019, 00h18

Tu ne devrais pas avoir besoin d’utiliser un custom reader ou une custom policy pour ce type de fichier.
Je pense que tu dois louper un truc. Normalement tu devrais avoir maximum 3, 4 lignes pour récupérer tout dans un tableau/writer. Ou alors l'API est mal foutue mais venant de Spring, ça serait étonnant.
Tu as jeté un oeil à https://howtodoinjava.com/spring-bat...essor-example/ ?

**ELSEV** · 30/10/2019, 09h52

J'ai utilisé cette technique en retournant un null afin de poursuivre les enregistrements entre les clients mais le traitement s'arrêtait.
De mémoire, je pense l'avoir fait dans l'ItemReader et non dans l'ItemProcessor comme dans l'exemple du lien que tu m'indiques

Merci pour cette nouvelle piste. Je ferais un test.

**ELSEV** · 30/10/2019, 11h18

J'ai fait le test, et toujours le même problème
En fait il me manque toujours l'instruction de dernier enregistrement pour transmettre de l'ItemProcessor, l'objet Client avec ses informations, vers l'ItemWriter

Comme je ne sais pas si c'est le dernier enregistrement d'un client, je continue à lire le fichier

Ci dessous le code de l'ItemProcessor

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
public class ClientProcessor implements ItemProcessor<BeanClient, BeanClient> {
 
	BeanClient beanClientPrevious;
 
	@Override
	public BeanClient process(BeanClient item) throws Exception {
		// Save le client precedent, complet ou non
		BeanClient beanClientFull = beanClientPrevious;
 
		if (beanClientPrevious == null) {
			// 1ier Client
			beanClientPrevious = item;
			return null;
		} else if (item.getClientName().equalsIgnoreCase(beanClientPrevious.getClientName())) {
			// Enregistrement suivant pour le meme client
			beanClientPrevious.addRecord(item);
			return null;
		} else if (item != null) {
			/* Rupture client */
				// Init nouveau client
			beanClientPrevious = item;
		}
 
		// Dans tous les cas 
		return beanClientFull;
	}

Ci dessous le code de LineMapper

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
 
public class OneClientLineMapper implements LineMapper<BeanClient> {
 
	@Override
	public BeanClient mapLine(String line, int lineNumber) throws Exception {
		BeanClient beanClient = new BeanClient(line);
		return beanClient;
	}

ci dessous les traces sur la console

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
 
ItemWriter = le client1 a : 1 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client2 a : 1 adresse, 1 numero de telephone, 1 facture
Job ended with status: exitCode=FAILED;exitDescription=

Il me manque les infos du client 3

Je pense que je devrais commencer par lire le nombre d'enregistrement du fichier,
puis comme dans le LineMapper j'ai le numéro de ligne je pourrais comparer et toper alors la dernier enregistrement comme fin de ligne dans mon objet client

**ELSEV** · 30/10/2019, 18h27

comme convenu, ci dessous ma solution en réalisant un count du fichier afin de le comparer avec la lecture de l'enregistrement en cours

Je réalise la lecture du nombre d'enregistrement du fichier au démarrage du Job, avec le JobExecutionListener. Mais on peut très bien le faire également dans une tasklet
Je place le nombre de ligne dans une Classe de constantes

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class ClientJobExecutionListener implements JobExecutionListener {
 
	private Resource inputResource;
 
	@Override
	public void beforeJob(JobExecution jobExecution) {
		try {
			CClientConstants.COUNT_LINE_CLIENT_CSV = Files.lines(Paths.get(inputResource.getFile().getPath())).count();
			System.out.println(ClientJobExecutionListener.class + " - Le fichier " + inputResource.getFilename() + " contient " + CClientConstants.COUNT_LINE_CLIENT_CSV + " enregistrements");
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
 
	@Override
	public void afterJob(JobExecution jobExecution) {
		// TODO Auto-generated method stub
	}
 
	public void setInputResource(Resource inputResource) {
		this.inputResource = inputResource;
	}
 
}

Dans le LineMapper je tope l'Objet client comme le dernier si c'est le dernier enregistrement du fichier

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
public class OneClientLineMapper implements LineMapper<BeanClient> {
 
	@Override
	public BeanClient mapLine(String line, int lineNumber) throws Exception {
		BeanClient beanClient = new BeanClient(line);
		// Topage du client si dernier enregistrement
		if (lineNumber == CClientConstants.COUNT_LINE_CLIENT_CSV) {
			beanClient.setLastClient(true);
		}
		return beanClient;
	}
}

Dans l'ItemProcessor, j'évalue si l'objet Client que je reçois et topé comme le dernier enregistrement

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public class ClientProcessor implements ItemProcessor<BeanClient, BeanClient> {
 
	BeanClient beanClientPrevious;
 
	@Override
	public BeanClient process(BeanClient item) throws Exception {
		// Save le client precedent, complet ou non
		BeanClient beanClientFull = beanClientPrevious;
 
		if (beanClientPrevious == null) {
			// 1ier Client
			beanClientPrevious = item;
			// Test si 1 seul enregistrement
			if (!item.isLastClient()) {
				return null;
			}
		} else if (item.getClientName().equalsIgnoreCase(beanClientPrevious.getClientName())) {
			// Enregistrement suivant pour le meme client
			beanClientPrevious.addRecord(item);
			// Test si dernier client
			if (!item.isLastClient()) {
				return null;
			}
		} else if (item != null) {
			/* Rupture client */
				// Init nouveau client
			beanClientPrevious = item;
		}
 
		// Dans tous les cas 
		return beanClientFull;
	}
}

Ci dessous l'ItemWriter pour exemple

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
public class ClientWriter implements ItemWriter<BeanClient> {
 
	public void write(List<? extends BeanClient> items) throws Exception {
		for (BeanClient beanClient : items) {
 
			System.out.println("ItemWriter = " + beanClient.toString());
		}
	}
}

Ci dessous la config du Job.xml, pour exemple

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
<?xml version="1.0" encoding="UTF-8"?> 
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:batch="http://www.springframework.org/schema/batch" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.springframework.org/schema/batch
		http://www.springframework.org/schema/batch/spring-batch-3.0.xsd
		http://www.springframework.org/schema/beans 
		http://www.springframework.org/schema/beans/spring-beans-4.3.xsd">
 
  <!-- Configuration Job pour lancer le batch en mmemoire -->	
  <bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
 
  <bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean" >	
  	<property name="transactionManager" ref="transactionManager" />
  </bean> 
 
  <bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository"/>
  </bean>
 
  <bean id="oneClientLineMapper" class="client.batch.OneClientLineMapper" />
 
 <bean id="clientJobExecutionListener" class="client.batch.ClientJobExecutionListener">
    <property name="inputResource" value="classpath:input/client.csv"/>
 </bean>
 
  <bean id="reader" class="org.springframework.batch.item.file.FlatFileItemReader">
  	<property name="resource" value="classpath:input/client.csv" />
  	<property name="lineMapper" ref="oneClientLineMapper"/>
  </bean>
 
  <bean id="clientProcessor" class="client.batch.ClientProcessor" />
  <bean id="writer" class="client.batch.ClientWriter" />
 
 <batch:job id="clientJob">
 
    <batch:step id="clientStep">
			<batch:tasklet>
				<batch:chunk reader="reader" processor="clientProcessor" writer="writer" commit-interval="1"/>
			</batch:tasklet>
	</batch:step>  
 
	<batch:listeners>
	   <batch:listener ref="clientJobExecutionListener"/>
	</batch:listeners>
  </batch:job> 
 
</beans>

et enfin les traces que j'obtiens sur la console

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
class client.batch.ClientJobExecutionListener - Le fichier client.csv contient 9 enregistrements
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

Merci Atha2 pour ta piste et tes retours
Elle me semble plus propre que d'utiliser un RecordSeparatorPolicy qui complexifie le process. Surtout comme tu le précisais j'en avais pas besoin avec la structure du fichier
Cela m'évite également d'ajouter une fin de ligne dans le fichier afin d'indiquer la fin du fichier

**atha2** · 30/10/2019, 18h54

Ah, je viens de tilter, tu ne peux pas forcément de permettre de lire un gros fichier (voir un flux) d'un seul coup avant de commencer le traitement (pb mémoire etc).
Dans mon esprit tu lisait tout et ensuite tu faisait ton traitement. Mais avec springbatch, ce n'est pas adapté.
En fait l'idée pour résoudre le problème c'est de faire du lookahead.
En me basant sur cette idée, j'ai codé le reader suivant :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
public class ClientRecordItemReader implements ItemReader<ClientRecords> {

    private String name;
    private BufferedReader reader;
    private String lookahead;

    public ClientRecordItemReader(Resource resource) {
        try {
            this.reader = new BufferedReader(new InputStreamReader(resource.getInputStream()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public ClientRecords read() throws IOException {
        ClientRecords currentRecord = null;
        if (lookahead == null) {
            lookahead = reader.readLine();
        }
        while (lookahead != null){
            System.out.println(">>>" + lookahead);
            String[] parts = lookahead.split(";");
            String lookaheadId = parts[0];
            if (currentRecord == null) {
                currentRecord = new ClientRecords(lookaheadId);
            }
            if ( ! lookaheadId.equals(currentRecord.getId())){
                System.out.println("lookaheadId="+lookaheadId + " currentRecord.getId="+currentRecord.getId());
                break;
            }

            List<String> record = new ArrayList<String>(Arrays.asList(parts));
            record.remove(0);
            currentRecord.addRecord(record);
            lookahead = reader.readLine();
        }
        System.out.println("return newly created :" + currentRecord);
        return currentRecord;
    }
    public ClientRecordItemReader name(String name) {
        this.name = name;

        return this;
    }
}

@Bean
public ItemReader<ClientRecords> clientReader() {
    return new ClientRecordItemReader(new ClassPathResource("client-data.csv"));
}

ça fonctionne chez moi avec ton fichier (sans aucun marqueur de fin de fichier).
Si besoin je peux t'uploader le projet.

Edit : Pendant que je rédigais mon message je vois que tu as finis par résoudre le problème. Ça fera toujours 2 solutions si quelqu'un retombe sur le même problème.

**ELSEV** · 31/10/2019, 10h02

Cool, merci pour ton exemple avec l'ItemReader
Oui ca fera 2 solutions possibles

Sinon, je pense pas que je lisais le fichier en one shot, mais par bloc de client avec le RecordSeparatorPolicy.
C'est vrai que ca peut porter à confusion mes sorties d'exemple que j'avais montré, mais en réalité j'avais ceci sur la console

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
 
postProcess = client1;adresse;adresse1||client1;telephone;telephone1;telephone1||client1;facture;facture1
LineMapper : client1;adresse;adresse1||client1;telephone;telephone1;telephone1||client1;facture;facture1
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
postProcess = client2;adresse;adresse1;adresse2||client2;telephone;telephone1||client2;facture;facture1
LineMapper : client2;adresse;adresse1;adresse2||client2;telephone;telephone1||client2;facture;facture1
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
postProcess = client3;adresse;adresse1||client3;telephone;telephone1||client3;facture;facture1;facture2
LineMapper : client3;adresse;adresse1||client3;telephone;telephone1||client3;facture;facture1;facture2
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture

L'ordre était bien ItemReader (avec le FlatFileItemReader, RecordSeparatorPolicy et LineMapper) puis directement l'ItemWriter
postProcess provient de la méthode de RecordSeparatorPolicy

Comme tu peux le voir, RecordSeparatorPolicy concatène en fait les lignes du fichier csv jusque ce que tu lui dis Stop via la méthode isEndOfRecord
Une fois Stop, le bloc partait vers LineMapper puis ItemWriter, puis le traitement reprenait dans RecordSeparatorPolicy avec le bloc d'enregistrement Client suivant
Inconvénient, j'étais obligé d'ajouter un séparateur "||" d'enregistrement pour les identifier et une fin de ligne pour identifier le dernier enregistrement

Dans le lineMaper, je décomposais mes blocs d'enregistrements dans mon Objet Client à l'aide du séparateur "||"

Bref j'avais peut être ajouté une complexité, mais j'ai appris à utiliser RecordSeparatorPolicy

Dans tous les cas, même si j'ai opté sur ton exemple avec l'ItemProcessor (plus logique avec la structure du fichier (comme tu me l'as fait remarqué et moins complexe), il me faut identifier une fin de ligne
. En l'occurrence, pour ma solution je fais un count du fichier que je compare ensuite avec le numéro de ligne dans le LineMapper

**ELSEV** · 31/10/2019, 11h37

J'ai fait un test avec ton ItemReader, apparemment à la dernière ligne après l'écriture des données du client3, il retournait dans l'ItemReader !!
J'ai dû spécifier un null pour lui dire que c'était terminé, du moins protéger la lecture d'un objet

ci dessous mon code, en reprenant ton exemple

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
public class ClientItemReader implements ItemReader<BeanClient> {
 
    private BufferedReader reader;
    private String currentLine;
 
    public ClientItemReader(Resource resource) {
        try {
			this.reader = new BufferedReader(new InputStreamReader(resource.getInputStream()));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
 
    @Override
	public BeanClient read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
		BeanClient currentBeanClient = null;
 
		// 1iere lecture
		if (currentLine == null) {
			currentLine = reader.readLine();
		}
 
		// Boucle sur les lignes
        while (currentLine != null) {
 
            System.out.println(">>>" + currentLine);
 
            if (currentBeanClient == null) {
            	currentBeanClient = new BeanClient(currentLine);
            } else {
            	BeanClient nextBeanClient = new BeanClient(currentLine);
 
            	if ( ! currentBeanClient.getClientName().equals(nextBeanClient.getClientName())){
 
	            	// Rupture Client
	                break;
	            }
 
            	// Enregistrement pour le meme client
                currentBeanClient.addRecord(currentLine);
            }
 
            // Lecture next line
            currentLine = reader.readLine();
        }
 
        if (currentBeanClient != null) {
	        System.out.println("return newly created :" + currentBeanClient.getClientName());
        }
 
        // /!\ currentBeanClient == null pour indiquer la fin de lecture du fichier
        return currentBeanClient;
	}
 
}

ce qui est interessant est que j'ai spécifié commit-interval="2" pour

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

<batch:chunk reader="clientReader" writer="writer" commit-interval="2"/>

et à ma console j'ai bien cette intervalle

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
>>>client1;adresse;adresse1
>>>client1;telephone;telephone1;telephone1
>>>client1;facture;facture1
>>>client2;adresse;adresse1;adresse2
return newly created :client1
>>>client2;adresse;adresse1;adresse2
>>>client2;telephone;telephone1
>>>client2;facture;facture1
>>>client3;adresse;adresse1
return newly created :client2
ItemWriter = le client1 a : 1 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
>>>client3;adresse;adresse1
>>>client3;telephone;telephone1
>>>client3;facture;facture1;facture2
return newly created :client3
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 1 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

**ELSEV** · 31/10/2019, 12h03

En utilisant la fonctionnalité de commit-interval, j'ai tout interêt, semble t-il à utiliser l'ItemReader pour gerer le fichier csv

ci dessous un exemple avec l'utilisation de ItemReader customisé, avec

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

<batch:chunk reader="clientReader" writer="writer" commit-interval="3"/>

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
>>>client1;adresse;adresse1
>>>client1;telephone;telephone1;telephone1
>>>client1;facture;facture1
>>>client2;adresse;adresse1;adresse2
return newly created :client1
>>>client2;adresse;adresse1;adresse2
>>>client2;telephone;telephone1
>>>client2;facture;facture1
>>>client3;adresse;adresse1
return newly created :client2
>>>client3;adresse;adresse1
>>>client3;telephone;telephone1
>>>client3;facture;facture1;facture2
return newly created :client3
ItemWriter = le client1 a : 1 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 1 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

L'intervalle d'écriture est bien géré

Tandis qu'en utilisant ItemProcessor avec FlatFileItemReader, avec

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
 
<batch:chunk reader="reader" processor="clientProcessor" writer="writer" commit-interval="5"/>

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

class client.batch.ClientJobExecutionListener - Le fichier client.csv contient 12 enregistrements
LineMapper ligne  = 1 - client1;adresse;adresse1
LineMapper ligne  = 2 - client1;telephone;telephone1;telephone1
LineMapper ligne  = 3 - client1;facture;facture1
LineMapper ligne  = 4 - client2;adresse;adresse1;adresse2
LineMapper ligne  = 5 - client2;telephone;telephone1
ItemProcessor = client1
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
LineMapper ligne  = 6 - client2;facture;facture1
LineMapper ligne  = 7 - client3;adresse;adresse1
LineMapper ligne  = 8 - client3;telephone;telephone1
LineMapper ligne  = 9 - client3;facture;facture1;facture2
LineMapper ligne  = 10 - client4;adresse;adresse1
ItemProcessor = client2
ItemProcessor = client3
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture
LineMapper ligne  = 11 - client4;telephone;telephone1
LineMapper ligne  = 12 - client4;facture;facture1;facture2
ItemProcessor = client4
ItemWriter = le client4 a : 1 adresse, 1 numero de telephone, 2 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

L'intervalle d'écriture n'est pas bien géré !
Les ItemWriter sont un peu éparpillé !!

D'où l'intérêt de partir sur ta solution avec l'ItemReader

Merci

**ELSEV** · 31/10/2019, 14h39

par contre, il semble que ItemReader, ayant une simple méthode read,
il est préconisé d'utiliser ItemStream qui lui possède 3 méthodes et permet de sauvegarder l'état du fichier en cas d'erreur et relance du traitement

ci dessous l'article :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
 
ItemReader has a single method read. Reader accepts a generic type and can read item of any type. The read method should return null when there is nothing more to read from the input source. Implementations should be stateful and should take care of maintaining the state.
 
 
The main purpose of ItemStream is to save state and restore from that state in case an error occurs. For this purpose, Item Stream utilizes three methods which must be implemented.
 
void open(ExecutionContext executionContext) throws ItemStreamException;
void update(ExecutionContext executionContext) throws ItemStreamException;
void close() throws ItemStreamException;
 
open is passed the previous execution context in case of restart or fresh execution context in case of first run. update saves the executionContext passed in the open method. close is used to release all the required resources. 
 
FlatFileItemReader is a implementation of ItemStreamReader as it should handle restarts gracefully.

ci dessous le lien des infos
https://examples.javacodegeeks.com/e...iters-example/

**atha2** · 31/10/2019, 16h12

Sur l'ordre des logs, même si c'est plus lisible, c'est n'est pas forcément plus efficace (à benchmarker pour savoir) ou plus correcte. Ma solution est paresseuse, on ne lit dans le fichier csv que quand le processor à besoin d''un nouveau clientItem. Je n'ai pas forcément cherché à optimiser mon code. Un système de pipeline avec des queues entre chaque étages pourrait être plus efficace, car contenant moins de pause. C'est ce qu'a l'air de faire ton autre solution (à voir qui gère les queues et le parallélisme apparaissant dans les logs).
Mais bon à part si tu as identifié/mesuré des limitations de performance avec une solution ou une autre, les optimisations ne sont pas nécessaires.

Sinon oui selon ton besoin, tu peux avoir besoin de reprendre là où tu en étais dans la lecture, auquel cas, sauver l'état du reader peut être utile. La doc officielle donne un exemple : https://docs.spring.io/spring-batch/...ReadersWriters.

Avec la solution du lookahead, il ne faudra pas oublier que la ligne lookahead n'est pas vraiment lue/prise en compte. En fait, il faut même stocker la ligne de reprise au moment où on créer un nouveau clientRecord/moment où on retourne un clientRecord complet.

**ELSEV** · 01/11/2019, 08h58

Merci de ta remarque Atha2,
pourquoi j'ai pensé que commit-interval était pour toute les étapes Reader, Process, et Writer pour terminer !
C'est sûr que c'est bien arrangeant d'avoir des logs bien propre de chaque étape, si c'est au détriment de la performance ...

Effectivement, en regardant la doc,
https://docs.spring.io/spring-batch/...html/step.html
En Figure 2. Chunk-oriented Processing, on comprend mieux le principe

L'ItemReader se charge de lire une à une les lignes du fichier et transmet de suite à ItemProcessor, et poursuit sa lecture sans se poser de question
L'ItemProcessor se charge quant à lui de traiter et de mettre de côté jusqu'à atteindre le commit-interval
Ce qui est sûr, qu'une fois dans le Writer, le bloc de données transmis de l'ItemProcessor et non seulement écrit mais aussi commité.
Avec le commit, Spring Batch ne retravaillera plus ces données commitées s'il y a reprise sur erreur. Parfait. Super.
D'autant plus qu'avec ma solution, la reprise sur erreur est implicitement géré par Spring Batch en ne gérant plus le Reader

Ce qui me gêne maintenant, si je tiens compte que ma solution est dans la logique de Spring Batch,
C'est que le commit-interval est fonction, semble t-il du reader, donc de ligne à ligne
Alors que j'aurais souhaité qu'il le fasse du côté du Writer, par bloc d'objet et donc par bloc de Client

Ex: si commit-interval="5", j'aurais souhaité qu'il écrit/commit 5 clients à la fois

si on regarde les logs de ma solution avec ItemProcessor

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class client.batch.ClientJobExecutionListener - Le fichier client.csv contient 12 enregistrements
LineMapper ligne  = 1 - client1;adresse;adresse1
LineMapper ligne  = 2 - client1;telephone;telephone1;telephone1
LineMapper ligne  = 3 - client1;facture;facture1
LineMapper ligne  = 4 - client2;adresse;adresse1;adresse2
LineMapper ligne  = 5 - client2;telephone;telephone1
ItemProcessor = client1
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
LineMapper ligne  = 6 - client2;facture;facture1
LineMapper ligne  = 7 - client3;adresse;adresse1
LineMapper ligne  = 8 - client3;telephone;telephone1
LineMapper ligne  = 9 - client3;facture;facture1;facture2
LineMapper ligne  = 10 - client4;adresse;adresse1
ItemProcessor = client2
ItemProcessor = client3
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture
LineMapper ligne  = 11 - client4;telephone;telephone1
LineMapper ligne  = 12 - client4;facture;facture1;facture2
ItemProcessor = client4
ItemWriter = le client4 a : 1 adresse, 1 numero de telephone, 2 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

on s'aperçoit qu'il commence à écrire/commité au bout de la 5ie ligne (du fichier) ... le Client1
pendant ce temps l'ItemReader est en train de lire le Client2
A la 10ie ligne, les client2 et 3 sont prêts à l'écriture/commit ... d'où l'écriture ...
et ainsi de suite avec les autre clients

les logs montrent bien que commit-interval est le reflet de la lecture et non de l'écriture
ennuyeux pour une structure de fichier comme le mien ... pour écrire/commité par Objet Client ...

En fait j'ai comme l'impression que le commit-interval perd son sens pour une structure de fichier comme le mien ...
D'où ma réflexion à ne pas l'utiliser, qui aurait pour valeur par défaut à 1
Ou à le surdimensionner mais au risque de mal le configurer

En effet, par exemple pour un client je peux avoir 500 lignes de données ... alors que d'autre je peux en avoir 5 ...
Je pense qu'il serait préférable de désactiver le commit-interval et de le forcer par moi-même ... si c'est possible .... et pas compliqué à gérer si on veut profiter de Spring Batch

**atha2** · 01/11/2019, 12h33

Effectivement le commit interval est utilisé de la façon suivante (citation de ton dernier lien) :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
 
List items = new Arraylist();
for(int i = 0; i < commitInterval; i++){
    Object item = itemReader.read()
    Object processedItem = itemProcessor.process(item);
    items.add(processedItem);
}
itemWriter.write(items);

Donc comme tu le dis c'est lié à la lecture. Je ne sais pas s'il est possible (ou même souhaitable) de remplacer ce code par une autre implémentation.
Si tu as toujours 3 lignes par client, au pire tu peux multiplier par 3 ton commit interval. Tu devrais avoir des logs clean et ordonnés de cette façon.
Mais il vaudrait mieux faire le * 3 quelque part dans le code (sinon la configuration serait trompeuse).

EDIT : sinon après avoir mieux compris pourquoi ton autre solution affichait des logs en détallé, je ne pense pas qu'une solution soit différente d'une autre d'un point de vue perf (aux micro optimisations près)

**ELSEV** · 02/11/2019, 09h00

En supprimant le commit-interval, Spring Batch m'informe qu'il est obligatoire et propose d'utiliser aussi 'chunk-completion-policy'

Le problème que je rencontre semble connu vu les cas, et ça me rassure
https://stackoverflow.com/questions/...-dynamic-value

et ce topic sur le forum avec un exemple
https://www.developpez.net/forums/d1...mmit-interval/

En utilisant cette méthode

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
<bean id="completionPolicy" class="org.springframework.batch.repeat.policy.DefaultResultCompletionPolicy"/>
<batch:chunk reader="reader" processor="clientProcessor" writer="writer" chunk-completion-policy="completionPolicy"/>

Ma solution semble se comporter comme un simple batch séquentiel : Lecture -> Logique -> Ecriture

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class client.batch.ClientJobExecutionListener - Le fichier client.csv contient 12 enregistrements
LineMapper ligne  = 1 - client1;adresse;adresse1
LineMapper ligne  = 2 - client1;telephone;telephone1;telephone1
LineMapper ligne  = 3 - client1;facture;facture1
LineMapper ligne  = 4 - client2;adresse;adresse1;adresse2
LineMapper ligne  = 5 - client2;telephone;telephone1
LineMapper ligne  = 6 - client2;facture;facture1
LineMapper ligne  = 7 - client3;adresse;adresse1
LineMapper ligne  = 8 - client3;telephone;telephone1
LineMapper ligne  = 9 - client3;facture;facture1;facture2
LineMapper ligne  = 10 - client4;adresse;adresse1
LineMapper ligne  = 11 - client4;telephone;telephone1
LineMapper ligne  = 12 - client4;facture;facture1;facture2
ItemProcessor = client1
ItemProcessor = client2
ItemProcessor = client3
ItemProcessor = client4
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture
ItemWriter = le client4 a : 1 adresse, 1 numero de telephone, 2 facture
Job ended with status: exitCode=COMPLETED;exitDescription=

Pour un simple fichier Client comme pour mes tests, ca passe. Sinon il y aura probablement un soucis de mémoire puisque le traitement semble tout lire le fichier en one shot, vu les traces. Il initialise d'abord tous les clients via le LineMapper depuis FlatFileItemReader, puis les traite dans l'ItemProcessor, et termine avec l'ItemWriter

Mais je pense qu'il est possible par cette méthode, de personnaliser le 'chunk-completion-policy'
En tout cas ça semble être une piste

**ELSEV** · 08/11/2019, 10h39

J'ai réussi à résoudre les 2 problèmes que je rencontrais
1 - le soucis de fin de ligne avec la structure de mon fichier, avec une lecture par client et non par ligne
2 - le commit-interval en relation avec le nombre de client et non par ligne

J'ai trouvé la solution dans cet exemple
https://keyholesoftware.com/2016/03/...egate-pattern/

Il s'agit de déléguer la lecture d'un fichier en ayant le contrôle de la lecture

Ci dessous mon code avec la ré-écriture de l'ItemReader : ClientDelegateItemReader

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
public class ClientDelegateItemReader implements ItemStreamReader<BeanClient> {
 
	private SingleItemPeekableItemReader<BeanClient> itemPeekable;
	private ItemReader<BeanClient> delegate;
 
	@BeforeStep
	public void beforeStep(StepExecution stepExecution) {
		itemPeekable = new SingleItemPeekableItemReader<BeanClient>();
		itemPeekable.setDelegate(delegate);
	}
 
	@Override
	public void open(ExecutionContext executionContext) throws ItemStreamException {
		itemPeekable.open(executionContext);
	}
 
	@Override
	public void update(ExecutionContext executionContext) throws ItemStreamException {
		itemPeekable.update(executionContext);
	}
 
	@Override
	public void close() throws ItemStreamException {
		itemPeekable.close();
	}
 
	@Override
	public BeanClient read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
		BeanClient currentBeanClient = null;
		BeanClient itemBeanClient;
		BeanClient nextItemBeanClient;
 
		while ( (itemBeanClient = itemPeekable.read()) != null ) {
			if (currentBeanClient == null) {
				currentBeanClient = itemBeanClient;
			} else {
				currentBeanClient.addRecord(itemBeanClient);
			}
 
			nextItemBeanClient = itemPeekable.peek();
			if (nextItemBeanClient == null || !currentBeanClient.getClientName().equals(nextItemBeanClient.getClientName())) {
				break;
			} 
		}
 
		return currentBeanClient;
	}
 
	public void setDelegate(ItemReader<BeanClient> delegate) {
		this.delegate = delegate;
	}
 
}

Ci dessous la config

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
  <bean id="flatFileReader" class="org.springframework.batch.item.file.FlatFileItemReader">
    <property name="resource" value="classpath:input/client.csv" />
    <property name="lineMapper" ref="oneClientLineMapper"/>
 
  </bean>
 
  <bean id="delegateFlatFileReader" class="client.batch.reader.ClientDelegateItemReader">
    <property name="delegate" ref="flatFileReader" />
  </bean>
 
 
  <bean id="clientProcessor" class="client.batch.ClientProcessor" />
  <bean id="writer" class="client.batch.ClientWriter" />
 
 <batch:job id="clientJob">
 
    <batch:step id="clientStep">
      <batch:tasklet>
        <batch:chunk reader="delegateFlatFileReader" writer="writer" commit-interval="2">
 
        </batch:chunk>
 
      </batch:tasklet>
  </batch:step>  
 
  </batch:job>

La ré-écriture de l'ItemReader avec l'implémentation de ItemStreamReader (avec les 3 méthodes : open, update,close) introduit SingleItemPeekableItemReader
Cette classe va me permettre de lire, grâce a sa méthode peek, la ligne suivante, et permettra ainsi de savoir si on est en fin de fichier.
La classe SingleItemPeekableItemReader possède également les 3 méthodes (open,update,close) .

La ré-écriture de l'ItemReader indroduit également la notion de délégation de service en utilisant en fait FlatFileItemReader
En déclarant le bean

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
<bean id="flatFileReader" class="org.springframework.batch.item.file.FlatFileItemReader">
    <property name="resource" value="classpath:input/client.csv" />
    <property name="lineMapper" ref="oneClientLineMapper"/>
 
  </bean>

Et en l'initialisant dans l'ItemReader Ré-écrit : ClientDelegateItemReader, par le biais d'un property que j'ai nommé delegate

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
 <bean id="delegateFlatFileReader" class="client.batch.reader.ClientDelegateItemReader">
    <property name="delegate" ref="flatFileReader" />
  </bean>

Et c'est ClientDelegateItemReader qui est déclaré dans le reader

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

 <batch:chunk reader="delegateFlatFileReader" writer="writer" commit-interval="2">

Et le tour est joué, pour lire une ligne en cours

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

itemPeekable.read()

Et vérifier si on est en fin de ligne

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

itemPeekable.peek()

avec fin de ligne si le retour de la méthode == null

Voici donc pour la solution de mon 1ier point

Pour le second,
le fait d'avoir le contrôle sur l'ItemReader, avec ClientDelegateItemReader,
j'indique que le retour d'un Item est un objet et non plus une ligne par défaut avec le FlatFileReader (dont j'utilise ses services)

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
@Override
public BeanClient read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
...
return currentBeanClient;

Avec commit-interval à 2

Code XML :

Sélectionner tout - Visualiser dans une fenêtre à part

<batch:chunk reader="delegateFlatFileReader" writer="writer" commit-interval="2">

J'ai l'écriture / commit tous les 2 clients

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
LineMapper ligne  = 1 - client1;adresse;adresse1
LineMapper ligne  = 2 - client1;telephone;telephone1;telephone1
LineMapper ligne  = 3 - client1;facture;facture1
LineMapper ligne  = 4 - client2;adresse;adresse1;adresse2
LineMapper ligne  = 5 - client2;telephone;telephone1
LineMapper ligne  = 6 - client2;facture;facture1
LineMapper ligne  = 7 - client3;adresse;adresse1
ItemWriter = le client1 a : 1 adresse, 2 numero de telephone, 1 facture
ItemWriter = le client2 a : 2 adresse, 1 numero de telephone, 1 facture
LineMapper ligne  = 8 - client3;telephone;telephone1
LineMapper ligne  = 9 - client3;facture;facture1;facture2
LineMapper ligne  = 10 - client4;adresse;adresse1
LineMapper ligne  = 11 - client4;telephone;telephone1
LineMapper ligne  = 12 - client4;facture;facture1;facture2
ItemWriter = le client3 a : 1 adresse, 1 numero de telephone, 2 facture
ItemWriter = le client4 a : 1 adresse, 1 numero de telephone, 2 facture

J'ai donc bien en sortie une écriture / commit tous les 2 clients avec une lecture ligne à ligne (via FlatItemReader) avec l'utilisation d'un LineMapper

[SPRINGBATCH] Lecture des enregistrements par client depuis un fichier [Batch]

Spring Java

Discussions similaires

Partager

Partager