Les retards à la SNCF Version Longue
(Facultatif) TP longue version
Pour ceux qui souhaitent approfondir le processus d'ETL en utilisant logstash
Analyse des données de l'Open Data de la SNCF
La première partie de ce TP portera sur l'analyse des données open data de la sncf portant sur les retards des lignes TGV en France à l'aide de logstash
Question
Rendez-vous dans le dossier logstash/conf/conf.d
Modifier le fichier 01-input.conf
Ce fichier contiendra la configuration permettant de lire le fichier regularite-mensuelle-tgv-short.csv placé dans le dossier /etc/logstash/files
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/01-input.conf2
input {3
file{4
path => [ "/etc/logstash/files/regularite-mensuelle-tgv-short.csv" ]
5
start_position => 'beginning'
6
sincedb_path => "/dev/null"
7
}8
}9
# /home/api04XX/logstash/conf/conf.d/01-input.conf
input {
file{
path => [ "/etc/logstash/files/regularite-mensuelle-tgv-short.csv" ]
start_position => 'beginning'
sincedb_path => "/dev/null"
}
}
Question
Modifier le fichier 02-filter.conf
Ce fichier contiendra la configuration permettant d'analyser un fichier csv avec les colonnes suivantes :
CTRL+C pour copier, CTRL+V pour coller
1
"date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite""date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/02-filter.conf2
filter {3
csv {4
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
5
separator => ";"
6
}7
}# /home/api04XX/logstash/conf/conf.d/02-filter.conf
filter {
csv {
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
separator => ";"
}
}Question
Modifier le fichier 03-output.conf
Ce fichier contiendra la configuration permettant de voir sur la console le résultat de l'analyse de logstash
Question
Lancer logstash et analyser le résultat
Solution
CTRL+C pour copier, CTRL+V pour coller
1
[api04XX@tpapi04 ~]$ start logstash
2
Starting api04XX_api04XX-logstash_1
3
Attaching to api04XX_api04XX-logstash_1
4
api04XX-logstash_1 | {5
api04XX-logstash_1 | "message" => [
6
api04XX-logstash_1 | [0] "2011-09;Atlantique;ANGOULEME;PARIS MONTPARNASSE;358;358;0;23;93.6\r"
7
api04XX-logstash_1 | ],
8
api04XX-logstash_1 | "@version" => "1",
9
api04XX-logstash_1 | "@timestamp" => "2016-01-17T13:26:57.562Z",
10
api04XX-logstash_1 | "host" => "api04XX-logstash",
11
api04XX-logstash_1 | "path" => "/etc/logstash/files/regularite-mensuelle-tgv-short.csv",
12
api04XX-logstash_1 | "date" => "2011-09",
13
api04XX-logstash_1 | "axe" => "Atlantique",
14
api04XX-logstash_1 | "depart" => "ANGOULEME",
15
api04XX-logstash_1 | "arrivee" => "PARIS MONTPARNASSE",
16
api04XX-logstash_1 | "trains_programmes" => "358",
17
api04XX-logstash_1 | "trains_circules" => "358",
18
api04XX-logstash_1 | "trains_annules" => "0",
19
api04XX-logstash_1 | "trains_retards" => "23",
20
api04XX-logstash_1 | "regularite" => "93.6"
21
api04XX-logstash_1 | }
[api04XX@tpapi04 ~]$ start logstash
Starting api04XX_api04XX-logstash_1
Attaching to api04XX_api04XX-logstash_1
api04XX-logstash_1 | {
api04XX-logstash_1 | "message" => [
api04XX-logstash_1 | [0] "2011-09;Atlantique;ANGOULEME;PARIS MONTPARNASSE;358;358;0;23;93.6\r"
api04XX-logstash_1 | ],
api04XX-logstash_1 | "@version" => "1",
api04XX-logstash_1 | "@timestamp" => "2016-01-17T13:26:57.562Z",
api04XX-logstash_1 | "host" => "api04XX-logstash",
api04XX-logstash_1 | "path" => "/etc/logstash/files/regularite-mensuelle-tgv-short.csv",
api04XX-logstash_1 | "date" => "2011-09",
api04XX-logstash_1 | "axe" => "Atlantique",
api04XX-logstash_1 | "depart" => "ANGOULEME",
api04XX-logstash_1 | "arrivee" => "PARIS MONTPARNASSE",
api04XX-logstash_1 | "trains_programmes" => "358",
api04XX-logstash_1 | "trains_circules" => "358",
api04XX-logstash_1 | "trains_annules" => "0",
api04XX-logstash_1 | "trains_retards" => "23",
api04XX-logstash_1 | "regularite" => "93.6"
api04XX-logstash_1 | }Problématique : On se rend compte que le fichier de la SNCF comporte une date qui n'est pas complète ce qui empêche logstash de l'analyser.
Question
Rajouter un filtre qui permettrait de compléter le champ date avec le premier jour du mois et de le convertir en timestamp afin de pouvoir par la suite l'analyser.
Indice
La conversion en timestamp s'effectue à l'aide du plugin date.
CTRL+C pour copier, CTRL+V pour coller
1
date {2
match => [ "champ", "YYYY-MM-dd" ] #On sélectionnne le pattern du champ afin d'éviter les erreurs de cast
3
timezone => "UTC"
4
}5
date {
match => [ "champ", "YYYY-MM-dd" ] #On sélectionnne le pattern du champ afin d'éviter les erreurs de cast
timezone => "UTC"
}
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/02-filter.conf2
filter {3
csv {4
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
5
separator => ";"
6
}7
8
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd9
mutate {10
replace => { "date" => "%{date}-01" }
11
}12
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.13
date {14
match => [ "date", "YYYY-MM-dd" ]
15
timezone => "UTC"
16
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp17
remove_field => ["date"]
18
}19
}# /home/api04XX/logstash/conf/conf.d/02-filter.conf
filter {
csv {
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
separator => ";"
}
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd
mutate {
replace => { "date" => "%{date}-01" }
}
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.
date {
match => [ "date", "YYYY-MM-dd" ]
timezone => "UTC"
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp
remove_field => ["date"]
}
}CTRL+C pour copier, CTRL+V pour coller
1
[api04XX@tpapi04 conf.d]$ start logstash
2
Starting api04XX_api04XX-logstash_1
3
Attaching to api04XX_api04XX-logstash_1
4
api04XX-logstash_1 | {5
api04XX-logstash_1 | "message" => [
6
api04XX-logstash_1 | [0] "2011-09;Atlantique;ANGOULEME;PARIS MONTPARNASSE;358;358;0;23;93.6\r"
7
api04XX-logstash_1 | ],
8
api04XX-logstash_1 | "@version" => "1",
9
api04XX-logstash_1 | "@timestamp" => "2011-09-01T00:00:00.000Z",
10
api04XX-logstash_1 | "host" => "api04XX-logstash",
11
api04XX-logstash_1 | "path" => "/etc/logstash/files/regularite-mensuelle-tgv-short.csv",
12
api04XX-logstash_1 | "axe" => "Atlantique",
13
api04XX-logstash_1 | "depart" => "ANGOULEME",
14
api04XX-logstash_1 | "arrivee" => "PARIS MONTPARNASSE",
15
api04XX-logstash_1 | "trains_programmes" => "358",
16
api04XX-logstash_1 | "trains_circules" => "358",
17
api04XX-logstash_1 | "trains_annules" => "0",
18
api04XX-logstash_1 | "trains_retards" => "23",
19
api04XX-logstash_1 | "regularite" => "93.6"
20
api04XX-logstash_1 | }
[api04XX@tpapi04 conf.d]$ start logstash
Starting api04XX_api04XX-logstash_1
Attaching to api04XX_api04XX-logstash_1
api04XX-logstash_1 | {
api04XX-logstash_1 | "message" => [
api04XX-logstash_1 | [0] "2011-09;Atlantique;ANGOULEME;PARIS MONTPARNASSE;358;358;0;23;93.6\r"
api04XX-logstash_1 | ],
api04XX-logstash_1 | "@version" => "1",
api04XX-logstash_1 | "@timestamp" => "2011-09-01T00:00:00.000Z",
api04XX-logstash_1 | "host" => "api04XX-logstash",
api04XX-logstash_1 | "path" => "/etc/logstash/files/regularite-mensuelle-tgv-short.csv",
api04XX-logstash_1 | "axe" => "Atlantique",
api04XX-logstash_1 | "depart" => "ANGOULEME",
api04XX-logstash_1 | "arrivee" => "PARIS MONTPARNASSE",
api04XX-logstash_1 | "trains_programmes" => "358",
api04XX-logstash_1 | "trains_circules" => "358",
api04XX-logstash_1 | "trains_annules" => "0",
api04XX-logstash_1 | "trains_retards" => "23",
api04XX-logstash_1 | "regularite" => "93.6"
api04XX-logstash_1 | }Question
Les valeurs numériques doivent aussi être castées par logstash pour être prise en compte par elasticsearch.
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/02-filter.conf2
filter {3
csv {4
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
5
separator => ";"
6
}7
8
# /home/api04XX/logstash/conf/conf.d/02-filter.conf9
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd10
mutate {11
replace => { "date" => "%{date}-01" }
12
convert => { "trains_programmes" => "integer" }
13
convert => { "trains_circules" => "integer" }
14
convert => { "trains_annules" => "integer" }
15
convert => { "trains_retards" => "integer" }
16
convert => { "regularite" => "float" }
17
}18
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.19
date {20
match => [ "date", "YYYY-MM-dd" ]
21
timezone => "UTC"
22
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp23
remove_field => ["date"]
24
}25
26
}# /home/api04XX/logstash/conf/conf.d/02-filter.conf
filter {
csv {
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
separator => ";"
}
# /home/api04XX/logstash/conf/conf.d/02-filter.conf
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd
mutate {
replace => { "date" => "%{date}-01" }
convert => { "trains_programmes" => "integer" }
convert => { "trains_circules" => "integer" }
convert => { "trains_annules" => "integer" }
convert => { "trains_retards" => "integer" }
convert => { "regularite" => "float" }
}
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.
date {
match => [ "date", "YYYY-MM-dd" ]
timezone => "UTC"
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp
remove_field => ["date"]
}
}Maintenant que les résultats correspondent à ce que l'on veut, on souhaite charger les données dans la base elasticsearch.
Question
Modifiez le fichier d'entrée pour utiliser le fichier regularite-mensuelle-tgv.csv
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/01-input.conf2
input {3
file{4
path => [ "/etc/logstash/files/regularite-mensuelle-tgv.csv" ]
5
start_position => 'beginning'
6
sincedb_path => "/dev/null"
7
}8
}# /home/api04XX/logstash/conf/conf.d/01-input.conf
input {
file{
path => [ "/etc/logstash/files/regularite-mensuelle-tgv.csv" ]
start_position => 'beginning'
sincedb_path => "/dev/null"
}
}Question
Modifiez le fichier de sortie pour utiliser une base elasticsearch
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/03-output.conf2
output {3
elasticsearch {4
host => "192.168.1.160"
5
cluster => "elastic.api04XX.docker"
6
port => "93XX"
7
index => "logstash-api04-%{+YYYY.MM.dd}"
8
protocol => "transport"
9
}10
}# /home/api04XX/logstash/conf/conf.d/03-output.conf
output {
elasticsearch {
host => "192.168.1.160"
cluster => "elastic.api04XX.docker"
port => "93XX"
index => "logstash-api04-%{+YYYY.MM.dd}"
protocol => "transport"
}
}Après l'ajout des données, il doit y avoir 100 shards actifs sur la base
CTRL+C pour copier, CTRL+V pour coller
1
curl localhost:92XX/_cluster/health?pretty=true
2
{3
"cluster_name" : "elastic.api0402.docker",
4
"status" : "green",
5
"timed_out" : false,
6
"number_of_nodes" : 1,
7
"number_of_data_nodes" : 1,
8
"active_primary_shards" : 100,
9
"active_shards" : 100,
10
"relocating_shards" : 0,
11
"initializing_shards" : 0,
12
"unassigned_shards" : 0,
13
"delayed_unassigned_shards" : 0,
14
"number_of_pending_tasks" : 0,
15
"number_of_in_flight_fetch" : 0
16
}
curl localhost:92XX/_cluster/health?pretty=true
{
"cluster_name" : "elastic.api0402.docker",
"status" : "green",
"timed_out" : false,
"number_of_nodes" : 1,
"number_of_data_nodes" : 1,
"active_primary_shards" : 100,
"active_shards" : 100,
"relocating_shards" : 0,
"initializing_shards" : 0,
"unassigned_shards" : 0,
"delayed_unassigned_shards" : 0,
"number_of_pending_tasks" : 0,
"number_of_in_flight_fetch" : 0
}Lors du lancement de logstash, vous pouvez voir des messages de la forme
CTRL+C pour copier, CTRL+V pour coller
1
{:timestamp=>"2016-01-17T14:20:50.351000+0000", :message=>"retrying failed action with response code: 429", :level=>:warn}{:timestamp=>"2016-01-17T14:20:50.351000+0000", :message=>"retrying failed action with response code: 429", :level=>:warn}Ce signifie que l'on charge "trop vite" l'elasticsearch et que logstash doit renvoyer les informations.
Question
Question subsidiaire : Dans le fichier de la SNCF, on voit qu'il y a le nom des gares de départ et d'arrivée.
Dans un second fichier opendata, la SNCF fournit le nom de ses points de vente ainsi que leurs coordonnées GPS. Elasticsearch est capable de stocker des points géo-point ce qui permet par la suite à Kibana d'afficher des cartes avec des points géographiques.
Il serait intéressant pour nous de pouvoir faire une sorte de jointure entre le fichier des localisations et le fichier des retards. Pour ça, il est possible d'écrire son propre plugin en ruby.
À partir du plugin situé dans le répertoire logstash/conf/conf.d/logstash/filter/localisation.rb ajoutez la configuration au fichier 02-filter.conf pour ajouter la localisation des villes dans les résultats de logstash.
Indice
CTRL+C pour copier, CTRL+V pour coller
1
require "logstash/filters/base"
2
require "logstash/namespace"
3
4
class LogStash::Filters::Localisation < LogStash::Filters::Base
5
6
7
# filter {8
# localisation { ... }9
# }10
config_name "localisation" #Nom du plugin
11
12
#Définition des variables13
config :database, :validate => :path # Le script va lire un fichier en entrée
14
config :source, :validate => :string, :required => true # Variable d'entrée avec une ville
15
config :target, :validate => :string, :default => 'location' # Variable de sortie pour stocker le tableau des positions GPS
16
config :locs
17
18
public19
def register
20
if @database.nil?
21
@database = LogStash::Environment.vendor_path("/etc/logstash/files/localisation.csv")
22
if !File.exists?(@database)
23
raise "la ligne 'database => ...' est obligatoire"
24
end25
end26
27
@locs = Hash.new
28
CSV.foreach(@database) do |row|
29
# Lecture des infos du fichier csv des villes et convertion en float30
# NB Elasticsearch analyse les géopoint stocké dans un tableau sous le forme longitude,latitude 31
@locs[row[0]] = [ row[2].to_f, row[1].to_f ] # { "lat" => row[1].to_f, "lon" => row[2].to_f }
32
end33
end34
35
36
public37
def filter(event)
38
# Ne retourne rien si la fonction ne renvoie rien 39
return unless filter?(event)
40
41
if event[@source]
42
loc = event[@source]
43
if @locs[loc]
44
event[@target] = @locs[loc]
45
else46
@logger.error("Localisation inconnue: ", {"loc" => loc})
47
end48
end49
# Les données récupérées sont renvoyées 50
filter_matched(event)
51
end52
end # class LogStash::Filters::Foo
53
require "logstash/filters/base"
require "logstash/namespace"
class LogStash::Filters::Localisation < LogStash::Filters::Base
# filter {
# localisation { ... }
# }
config_name "localisation" #Nom du plugin
#Définition des variables
config :database, :validate => :path # Le script va lire un fichier en entrée
config :source, :validate => :string, :required => true # Variable d'entrée avec une ville
config :target, :validate => :string, :default => 'location' # Variable de sortie pour stocker le tableau des positions GPS
config :locs
public
def register
if @database.nil?
@database = LogStash::Environment.vendor_path("/etc/logstash/files/localisation.csv")
if !File.exists?(@database)
raise "la ligne 'database => ...' est obligatoire"
end
end
@locs = Hash.new
CSV.foreach(@database) do |row|
# Lecture des infos du fichier csv des villes et convertion en float
# NB Elasticsearch analyse les géopoint stocké dans un tableau sous le forme longitude,latitude
@locs[row[0]] = [ row[2].to_f, row[1].to_f ] # { "lat" => row[1].to_f, "lon" => row[2].to_f }
end
end
public
def filter(event)
# Ne retourne rien si la fonction ne renvoie rien
return unless filter?(event)
if event[@source]
loc = event[@source]
if @locs[loc]
event[@target] = @locs[loc]
else
@logger.error("Localisation inconnue: ", {"loc" => loc})
end
end
# Les données récupérées sont renvoyées
filter_matched(event)
end
end # class LogStash::Filters::Foo
Solution
CTRL+C pour copier, CTRL+V pour coller
1
# /home/api04XX/logstash/conf/conf.d/02-filter.conf2
filter {3
csv {4
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
5
separator => ";"
6
}7
8
# /home/api04XX/logstash/conf/conf.d/02-filter.conf9
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd10
mutate {11
replace => { "date" => "%{date}-01" }
12
convert => { "trains_programmes" => "integer" }
13
convert => { "trains_circules" => "integer" }
14
convert => { "trains_annules" => "integer" }
15
convert => { "trains_retards" => "integer" }
16
convert => { "regularite" => "float" }
17
}18
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.19
date {20
match => [ "date", "YYYY-MM-dd" ]
21
timezone => "UTC"
22
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp23
remove_field => ["date"]
24
}25
26
#Rajoute les coordonnées GPS des villes pour Kibana27
localisation {28
database => "/etc/logstash/files/localisation.csv"
29
source => "depart"
30
target => "departGPS"
31
}32
localisation {33
database => "/etc/logstash/files/localisation.csv"
34
source => "arrivee"
35
target => "arriveeGPS"
36
}37
}# /home/api04XX/logstash/conf/conf.d/02-filter.conf
filter {
csv {
columns => ["date","axe","depart","arrivee","trains_programmes","trains_circules","trains_annules","trains_retards","regularite"]
separator => ";"
}
# /home/api04XX/logstash/conf/conf.d/02-filter.conf
#Modification du champ date qui est de la forme yyyy-mm vers yyyy-mm-dd
mutate {
replace => { "date" => "%{date}-01" }
convert => { "trains_programmes" => "integer" }
convert => { "trains_circules" => "integer" }
convert => { "trains_annules" => "integer" }
convert => { "trains_retards" => "integer" }
convert => { "regularite" => "float" }
}
#cast du champ date que l'on vient de modifier vers un timestamp reconnu par elasticsearch.
date {
match => [ "date", "YYYY-MM-dd" ]
timezone => "UTC"
#Si la conversion a réussi, on supprime le champ date. Par défaut le résultat de la fonction date est envoyé au champ @timestamp
remove_field => ["date"]
}
#Rajoute les coordonnées GPS des villes pour Kibana
localisation {
database => "/etc/logstash/files/localisation.csv"
source => "depart"
target => "departGPS"
}
localisation {
database => "/etc/logstash/files/localisation.csv"
source => "arrivee"
target => "arriveeGPS"
}
}Lorsque l'on ajoute des geopoints, il faut spécifier un mapping spécifique pour caster les coordonnées.
CTRL+C pour copier, CTRL+V pour coller
1
resetES
2
3
curl -XPUT 'localhost:92XX/_template/logstash-api04' -d @/home/api04XX/logstash/conf/files/mappingGPS
4
5
start logstash
resetES curl -XPUT 'localhost:92XX/_template/logstash-api04' -d @/home/api04XX/logstash/conf/files/mappingGPS start logstash
