o_0 | public enum {java,logiciels libres, gnu/linux,elaticsearch,... }

Créer un connecteur KAFKA CONNECT

9 septembre 201614 septembre 2016 littlewing Laisser un commentaire

Après avoir essayé le connecteur de confluent kafka-connect-jdbc, je me suis aperçu qu’il y avait encore quelques bugs qui le rendaient inutilisable dans mon environnement ( notamment avec les champs numériques).

J’ai donc décidé de créer mon propre connecteur. Certes celui-ci sera moins générique mais il correspondra à mon besoin ( ou pas …).

API utilisées

JAVA8
API KAFKA CONNECT

Configuration maven

Cette configuration me permet de créer un fat jar avec les dépendances nécessaires à la bonne exécution du connecteur.

Voici mon fichier pom.xml

  <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>info.touret.myconnect</groupId>
    <artifactId>myconnect</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>myconnect</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <kafka.version>0.10.0.0</kafka.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>LATEST</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.mockito</groupId>
            <artifactId>mockito-all</artifactId>
            <version>LATEST</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.powermock</groupId>
            <artifactId>powermock-api-mockito</artifactId>
            <version>1.6.5</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.powermock</groupId>
            <artifactId>powermock-module-junit4</artifactId>
            <version>1.6.5</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.oracle.weblogic</groupId>
            <artifactId>ojdbc7</artifactId>
            <version>12.1.3-0-0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-runtime</artifactId>
            <version>${kafka.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>connect-api</artifactId>
            <version>${kafka.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.21</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.21</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.typesafe</groupId>
            <artifactId>config</artifactId>
            <version>1.3.0</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>1.7.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <shadedArtifactAttached>true</shadedArtifactAttached>
                            <shadedClassifierName>allinone</shadedClassifierName>
                            <artifactSet>
                                <includes>
                                    <include>*:*</include>
                                </includes>
                            </artifactSet>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>reference.conf</resource>
                                </transformer>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <manifestEntries>
                                        <Main-Class></Main-Class>
                                    </manifestEntries>
                                </transformer>
                            </transformers>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>


        </plugins>
    </build>
</project>

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>info.touret.myconnect</groupId>

<artifactId>myconnect</artifactId>

<version>1.0-SNAPSHOT</version>

<name>myconnect</name>

<url>http://maven.apache.org</url>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<kafka.version>0.10.0.0</kafka.version>

</properties>

<groupId>junit</groupId>

<artifactId>junit</artifactId>

<version>LATEST</version>

</dependency>

<groupId>org.mockito</groupId>

<artifactId>mockito-all</artifactId>

<version>LATEST</version>

</dependency>

<groupId>org.powermock</groupId>

<artifactId>powermock-api-mockito</artifactId>

</dependency>

<groupId>org.powermock</groupId>

<artifactId>powermock-module-junit4</artifactId>

</dependency>

<groupId>com.oracle.weblogic</groupId>

<artifactId>ojdbc7</artifactId>

</dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>connect-runtime</artifactId>

<version>${kafka.version}</version>

<scope>provided</scope>

</dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>connect-api</artifactId>

<version>${kafka.version}</version>

<scope>provided</scope>

</dependency>

<scope>provided</scope>

</dependency>

<scope>provided</scope>

</dependency>

<groupId>com.typesafe</groupId>

<artifactId>config</artifactId>

</dependency>

</dependencies>

<build>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

</configuration>

</plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-shade-plugin</artifactId>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

<shadedClassifierName>allinone</shadedClassifierName>

</includes>

</artifactSet>

<transformer

implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">

<resource>reference.conf</resource>

</transformer>

<transformer

implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">

<Main-Class></Main-Class>

</manifestEntries>

</transformer>

</transformers>

</excludes>

</filter>

</filters>

</configuration>

</execution>

</executions>

</plugin>

</plugins>

</build>

</project>

Développement

Comme le dit le guide de développement, il faut créer a minima une classe héritant de la classe SourceConnector et une classe héritant de SourceTask.

MySourceConnector

Cette classe permet la récupération de la configuration du connecteur.

public class MySourceConnector extends SourceConnector {

    public static final String TOPIC_CONFIG = "topic";

    private static Logger logger = LoggerFactory.getLogger(MySourceConnector.class);

    public String version() {
        return AppInfoParser.getVersion();
    }

    public void start(Map&lt;String, String&gt; map) {
    }

    public Class&lt;? extends Task&gt; taskClass() {
        return MySourceTask.class;
    }

    public List&lt;Map&lt;String, String&gt;&gt; taskConfigs(int maxTasks) {
        ArrayList&lt;Map&lt;String, String&gt;&gt; configs = new ArrayList&lt;&gt;();
        Map&lt;String, String&gt; map = new HashMap&lt;&gt;();
        map.put(TOPIC_CONFIG, "montopic");
        configs.add(map);
        return configs;
    }

    public void stop() {

    }

    public ConfigDef config() {
        return new ConfigDef().define(TOPIC_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Kafka Topic");
    }
}

public class MySourceConnector extends SourceConnector {

public static final String TOPIC_CONFIG = "topic";

private static Logger logger = LoggerFactory.getLogger(MySourceConnector.class);

public String version() {

return AppInfoParser.getVersion();

}

public void start(Map<String, String> map) {

}

public Class<? extends Task> taskClass() {

return MySourceTask.class;

}

public List<Map<String, String>> taskConfigs(int maxTasks) {

ArrayList<Map<String, String>> configs = new ArrayList<>();

Map<String, String> map = new HashMap<>();

map.put(TOPIC_CONFIG, "montopic");

configs.add(map);

return configs;

}

public void stop() {

}

public ConfigDef config() {

return new ConfigDef().define(TOPIC_CONFIG, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Kafka Topic");

}

MySourceTask

Cette classe gère l’exécution de l’extraction et chargement dans KAFKA. Elle permet dans la méthode start() de démarrer le connecteur et de lancer les ressources (connexions JDBC).

Comme le connecteur standard, je m’appuie sur un champ de type Timestamp. celui -ci me permet de créer un offset et de faire un parcours incrémental de mes résultats .

public class MySourceTask extends SourceTask {

    private String topic = null;
    private Logger logger = LoggerFactory.getLogger(MySourceTask.class);
    private Connection connection;
    private PreparedStatement preparedStatement;

    public void start(Map<String, String> props) {
        topic = props.get(MySourceConnector.TOPIC_CONFIG);
        logger.warn("Starting Task ");
        try {
            connection = createConnection(); // connexion jdbc classique
            preparedStatement = connection.prepareStatement("select *from mytable");
        } catch (SQLException e) {
            logger.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    public List<SourceRecord> poll() throws InterruptedException {
        logger.warn("Polling");
        List<SourceRecord> results = new LinkedList<>();
        try {
// recuperation de la partition
            Map sourcePartition = Collections.singletonMap("table", "mytable"));
// recuperation du dernier offset
            final Timestamp lastRecordedOffset = getLastRecordedOffset(sourcePartition);
            preparedStatement.setTimestamp(1,lastRecordedOffset);
            ResultSet resultSet = preparedStatement.executeQuery();
            while (resultSet.next()) {
// recuperation du champ de type timestamp permettant le chargement incremental
                Timestamp timestamp = resultSet.getTimestamp(18);
// extraction de la ligne (oui je sais c'est bourrin)
                String line = new StringBuilder().append(resultSet.getString(1)).append(resultSet.getString(2)).append(timestamp).toString();
je cree un nouveau offset 
                Map sourceOffSet = Collections.singletonMap("position", new Long(timestamp.getTime()));
                results.add(new SourceRecord(sourcePartition, sourceOffSet, topic, Schema.STRING_SCHEMA, line));
            }
        } catch (SQLException e) {
            throw new InterruptedException(e.getMessage());
        }
        return results;
    }
/**
 * extraction du timestamp du dernier champ extrait. Par defaut, je positionne l'EPOCH
 */
    private Timestamp getLastRecordedOffset(Map<String,Object> partition) {
        Map<String,Object> offset = context.offsetStorageReader().offset(partition);
        Timestamp lastRecordedOffset = Timestamp.from(EPOCH);
        if(offset !=null){
            lastRecordedOffset = new Timestamp((Long)offset.getOrDefault("position",Timestamp.from(EPOCH)));
        }
        return lastRecordedOffset;
    }

    public void stop() {
        try {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (preparedStatement != null) {
                connection.close();
            }
        } catch (SQLException e) {
            logger.error(e.getMessage(), e);
        }
    }

    public String version() {
        return new MySourceConnector().version();
    }
}

public class MySourceTask extends SourceTask {

private String topic = null;

private Logger logger = LoggerFactory.getLogger(MySourceTask.class);

private Connection connection;

private PreparedStatement preparedStatement;

public void start(Map<String, String> props) {

topic = props.get(MySourceConnector.TOPIC_CONFIG);

logger.warn("Starting Task ");

try {

connection = createConnection(); // connexion jdbc classique

preparedStatement = connection.prepareStatement("select *from mytable");

} catch (SQLException e) {

logger.error(e.getMessage(), e);

throw new RuntimeException(e);

}

public List<SourceRecord> poll() throws InterruptedException {

logger.warn("Polling");

List<SourceRecord> results = new LinkedList<>();

try {

// recuperation de la partition

Map sourcePartition = Collections.singletonMap("table", "mytable"));

// recuperation du dernier offset

final Timestamp lastRecordedOffset = getLastRecordedOffset(sourcePartition);

preparedStatement.setTimestamp(1,lastRecordedOffset);

ResultSet resultSet = preparedStatement.executeQuery();

while (resultSet.next()) {

// recuperation du champ de type timestamp permettant le chargement incremental

Timestamp timestamp = resultSet.getTimestamp(18);

// extraction de la ligne (oui je sais c'est bourrin)

String line = new StringBuilder().append(resultSet.getString(1)).append(resultSet.getString(2)).append(timestamp).toString();

je cree un nouveau offset

Map sourceOffSet = Collections.singletonMap("position", new Long(timestamp.getTime()));

results.add(new SourceRecord(sourcePartition, sourceOffSet, topic, Schema.STRING_SCHEMA, line));

}

} catch (SQLException e) {

throw new InterruptedException(e.getMessage());

}

return results;

}

/**

* extraction du timestamp du dernier champ extrait. Par defaut, je positionne l'EPOCH

private Timestamp getLastRecordedOffset(Map<String,Object> partition) {

Map<String,Object> offset = context.offsetStorageReader().offset(partition);

Timestamp lastRecordedOffset = Timestamp.from(EPOCH);

if(offset !=null){

lastRecordedOffset = new Timestamp((Long)offset.getOrDefault("position",Timestamp.from(EPOCH)));

}

return lastRecordedOffset;

}

public void stop() {

try {

if (preparedStatement != null) {

preparedStatement.close();

}

if (preparedStatement != null) {

connection.close();

}

} catch (SQLException e) {

logger.error(e.getMessage(), e);

}

public String version() {

return new MySourceConnector().version();

}

Configuration nécessaire à l’exécution du plugin

Il faudra créer également un fichier properties contenant les informations suivantes :

name=my-connector-source
connector.class=info.touret.MySourceConnector

1 2	name=my-connector-source connector.class=info.touret.MySourceConnector

Exécution

De la même manière que pour le connecteur standard…

Conclusion

Voila le squelette de mon connecteur crée. Pour l’instant les données sont sérialisées de manière un peu brutale. La prochaine étape sera de les mettre au format JSON. La suite dans un prochain numéro…

Mettre en place KAFKA CONNECT et KAFKA pour faire du change data capture (CDC)

30 août 20165 septembre 2016 littlewing Laisser un commentaire

Je suis en train de monter un moteur d’analyse permettant d’analyser les logs d’un ESB.

L’outil standard stockes les évènements dans une base de données.

Afin d’être le moins intrusif possible dans le code des médiations, j’ai choisi de mettre en place un mécanisme de change data capture (CDC) avec KAFKA CONNECT et KAFKA. Pour ce faire j’ai déployé la solution confluent. Celle-ci permet entre autres d’avoir des connecteurs JDBC permettant la connexion aux bases de données relationnelles.

KAFKA est un système de messagerie distribué crée initialement par LinkedIn. Kafka connect fournit une interface en entrée et en sortie de ce dernier.

Si vous voulez plus d’informations, vous pouvez aller du coté de la présentation du devox 2016

Installation

J’ai utilisé la procédure décrite ici

Démarrage des différents démons

./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties 
./bin/kafka-server-start ./etc/kafka/server.properties 
./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties 
./bin/kafka-rest-start ./etc/kafka-rest/kafka-rest.properties

./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties

./bin/kafka-server-start ./etc/kafka/server.properties

./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties

./bin/kafka-rest-start ./etc/kafka-rest/kafka-rest.properties

Configuration du connecteur JDBC pour KAFKA CONNECT

Il faut tout d’abord copier le driver jdbc dans le répertoire share/java/kafka-connect-jdbc/

Ensuite, il faut créer un fichier de configuration (ex. esb-logs.properties )

name=esb-log-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10
connection.url=URL_JDBC
table.whitelist=ma_table
mode=timestamp
timestamp.column.name=jobstart
topic.prefix=prefix

name=esb-log-source

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector

tasks.max=10

connection.url=URL_JDBC

table.whitelist=ma_table

mode=timestamp

timestamp.column.name=jobstart

topic.prefix=prefix

Les contraintes de KAFKA CONNECT sont les suivantes

Si on veut faire des extractions incrémentales ( ce qui est mon cas) , il faut soit un élément de type TIMESTAMP qui ne puisse pas être NULL, soit une PK auto-incrémentée.
Les requêtes doivent être relativement assez simples. Personnellement, je préfère utiliser des vues qui me retournent l’exécution des requêtes SQL.

Exécution

Lancer la commande

 ./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/esb-logs.properties

1	./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/esb-logs.properties

Pour vérifier que tout est bien dans le topic KAFKA

./bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic prefix- --from-beginning

1	./bin/kafka-avro-console-consumer --zookeeper localhost:2181 --topic prefix- --from-beginning

Conclusion

Pour l’instant, j’ai réussi à extraire des données de ma base de données vers un broker KAFKA. La suite me permettra de les traiter en mode streaming dans SPARK.

Utiliser SBT derrière un proxy d’entreprise

3 juin 2016 littlewing Laisser un commentaire

J’essaye depuis quelques temps de faire fonctionner SBT derrière un proxy d’entreprise ( avec authentification) . Vu que je ne veux pas polluer le serveur NEXUS existant, j’ai essayé de tout faire un local …. et là c’est le drame.

Jusqu’à présent, j’utilisais MAVEN et le plugin SCALA pour répondre aux besoins du BUILD. J’ai néanmoins trouvé une autre solution, tout aussi sale…. Installer un serveur NEXUS sur le poste de travail et s’en servir comme proxy « intelligent ».

J’ai fait la configuration suivante :

Je ne vais pas décrire ici l’installation de nexus tellement c’est simple.

Dans la configuration de nexus, le proxy se configure assez simplement dans la partie server

J’ai ajouté les virtual repositories suivants :

ivy-releases-1: https://dl.bintray.com/sbt/sbt-plugin-releases/
ivy-releases-2: https://dl.bintray.com/typesafe/ivy-releases/
mvn-typesafe-release-1: http://repo.typesafe.com/typesafe/maven-releases/
mvn-typesafe-release-2: http://repo.typesafe.com/typesafe/releases/

et crée deux groupes l’un pour les repo ivy (ivy-releases) et l’autre pour maven (mvn-releases)

Enfin, comme l’indique la documentation, j’ai configuré le fichier ~.sbt/repositories de la manière suivante :

[repositories]
  local
  my-ivy-proxy-releases: http://127.0.0.1:8081/nexus/content/groups/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext]
  my-maven-proxy-releases: http://127.0.0.1:8081/nexus/content/groups/mvn-releases/

[repositories]

local

my-ivy-proxy-releases: http://127.0.0.1:8081/nexus/content/groups/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext]

my-maven-proxy-releases: http://127.0.0.1:8081/nexus/content/groups/mvn-releases/

Enfin, lorsque vous exécutez sbt, celui-ci utilisera automatiquement votre proxy NEXUS qui lui est capable de discuter avec un proxy d’entreprise.

En espérant que ça serve à d’autres…

Normaliser une adresse avec ElasticSearch et la base adresse nationale

31 mai 2016 littlewing Laisser un commentaire

Il y a fort fort longtemps, j’intégrais la base adresse nationale dans elasticsearch via logstash. Je ne suis pas allé plus loin faute de temps et peut-être d’envie.

Que peut-on faire avec ces quelques gigas de données me direz vous?

Faire des recherches pour avoir les coordonnées géographiques d’une adresse donnée
Faire une recherche pour avoir l’adresse normalisée
Faire un rapprochement avec d’autres données (avec Spark par exemple)
Sans doute plein d’autres use cases

Je vais m’attarder sur le deuxième point. A quoi ça sert ? et bien à avoir une adresse « propre » et utilisable par un système d’information (ex. ce que font les impôts, ou amazon). Il existe quelques solutions propriétaires qui réalisent ceci et sont assez chers. Je me suis donc mis dans la tête de le faire via elasticsearch.

Rappels des épisodes précédents

Voici le schéma de l’architecture

Pré-requis

Installer Elasticsearch > 2.3
Installer Logstash
Récupérer l’archive de la base adresse nationale

Configuration Logstash

Voici ma configuration Logstash

input {
  file {
    codec =&gt;plain{
    charset =&gt; "UTF-8"
    }
    path =&gt; ["./data/*.csv"]
    sincedb_path =&gt; "logstash/.sincedb_path"
    start_position =&gt; "beginning"
  }
}

filter {
	if [message] =~ /^"id";"nom_voie"/ {
	drop { }
	}
 	csv {
    columns =&gt; ["id","nom_voie","id_fantoir","numero","rep","code_insee","code_post","alias","nom_ld","nom_afnor","libelle_acheminement","x","y","lon","lat","nom_commune"]
    separator =&gt; ";"
	source =&gt; message
  }

  mutate {
    rename =&gt; [ "lat", "[location][lat]", "lon", "[location][lon]" ]
  }
   mutate {
    convert =&gt; [ "[location][lat]", "float" ]
    convert =&gt; [ "[location][lon]", "float" ]
    convert =&gt; [ "[lon]", "float" ]
    convert =&gt; [ "[lat]", "float" ]
    convert =&gt; ["[location]","float"]
    }
}


output {
        stdout{
        codec =&gt; "json"
        }
    elasticsearch {
    hosts =&gt; ["localhost"]
    index =&gt; "ban"
    document_type =&gt; "address"
    workers =&gt; 8
    document_id =&gt; "%{id}"
  }
}

input {

file {

codec =>plain{

charset => "UTF-8"

}

path => ["./data/*.csv"]

sincedb_path => "logstash/.sincedb_path"

start_position => "beginning"

}

filter {

if [message] =~ /^"id";"nom_voie"/ {

drop { }

}

csv {

columns => ["id","nom_voie","id_fantoir","numero","rep","code_insee","code_post","alias","nom_ld","nom_afnor","libelle_acheminement","x","y","lon","lat","nom_commune"]

separator => ";"

source => message

}

mutate {

rename => [ "lat", "[location][lat]", "lon", "[location][lon]" ]

}

mutate {

convert => [ "[location][lat]", "float" ]

convert => [ "[location][lon]", "float" ]

convert => [ "[lon]", "float" ]

convert => [ "[lat]", "float" ]

convert => ["[location]","float"]

}

output {

stdout{

codec => "json"

}

elasticsearch {

hosts => ["localhost"]

index => "ban"

document_type => "address"

workers => 8

document_id => "%{id}"

}

J’ai configuré LOGSTASH pour réaliser les actions suivantes :

extraction des données d’un fichier
suppression de la ligne d’en-tête
gestion de la localisation avec un champ de type ‘geo_point’.

C’est assez simple (pour l’instant)

Mapping ELASTICSEARCH

Analyzers

J’ai configuré les analyzers de la manière suivante:

"analysis": {
      "analyzer": {
        "ngram_analyzer": {
          "type": "custom",
          "tokenizer": "whitespace",
          "filter": [
            "lowercase",
            "asciifolding",
            "ngram_filter"
          ]
        },
        "whitespace_analyzer": {
          "type": "custom",
          "tokenizer": "whitespace",
          "filter": [
            "lowercase",
            "asciifolding"
          ]
        }
      },
      "filter": {
        "ngram_filter": {
          "type": "nGram",
          "min_gram": 3,
          "max_gram": 15,
          "token_chars": [
            "letter",
            "digit",
            "punctuation",
            "symbol"
          ]
        }
      }
    }

"analysis": {

"analyzer": {

"ngram_analyzer": {

"type": "custom",

"tokenizer": "whitespace",

"filter": [

"lowercase",

"asciifolding",

"ngram_filter"

]

"whitespace_analyzer": {

"type": "custom",

"tokenizer": "whitespace",

"filter": [

"lowercase",

"asciifolding"

]

}

"filter": {

"ngram_filter": {

"type": "nGram",

"min_gram": 3,

"max_gram": 15,

"token_chars": [

"letter",

"digit",

"punctuation",

"symbol"

]

}

J’ analyse tous les mots (whitespace tokenizer) en minuscule en appliquant un filtre (ngram) permet de rechercher par caractère (ex. imp au lieu de impasse).

La c’est super couteux en espace disque et impose un plus gros traitement lors de l’insertion mais permet d’ alléger le temps de traitement des requêtes ( le gros du travail est fait lors du chargement).

Mapping

La je n’ai pas fait grand chose de particulier si ce n’est le typage des différents champs (ex. la localisation )

   "location": {
               "type": "geo_point"
       
          }

"location": {

"type": "geo_point"

}

Création de l’index

Dans sense ou via cUrl, lancer la commande suivante :

PUT /ban{...}

1	PUT /ban{...}

Pour le mapping complet, voir sur mon compte github

Chargement

voir mon article précédent

Interrogation

Maintenant je peux interroger mon index

Imaginons que je fasse saisir le code postal, la ville et l’adresse et que j’interroge elasticsearch pour obtenir une adresse normalisée

#/GET /ban/address/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "rue",
            "type": "best_fields",
            "fields": [
              "nom_ld",
              "nom_afnor"
            ]
          }
        },
        {
          "match": {
            "nom_commune": {
              "query": "ville"
            }
          }
        }
      ],
      "filter": [
        {
          "term": {
            "code_post": "00000"
          }
        }
      ]
    }
  }
}

#/GET /ban/address/_search

{

"query": {

"bool": {

"must": [

{

"multi_match": {

"query": "rue",

"type": "best_fields",

"fields": [

"nom_ld",

"nom_afnor"

]

}

{

"match": {

"nom_commune": {

"query": "ville"

}

"filter": [

{

"term": {

"code_post": "00000"

}

]

}

Conclusion

Ce n’est qu’un début. Ce n’est sans doute pas encore très performant, si vous avez des remarques, n’hésitez pas. En tout cas, les briques et outils décrits ci-dessus permettent de normaliser des adresses a minima avec des logiciels libres, ce qui n’est pas rien.

Les sources sont disponibles sur GITHUB. Je ne pense pas que je vais en faire un projet à proprement parler mais plutôt d’une boîte à outils pour manipuler ce genre de données. J’y ajouterai sans doute quelques recherches géospatiales et peut être l’ intégration de graph.

Quoi de neuf du coté du front-end

13 mai 2016 littlewing Laisser un commentaire

Depuis le début de l’année je suis en train de faire une spécialisation Coursera : Full Stack Developer.

Pourquoi me direz-vous ? Tout d’abord, j’ai principalement travaillé sur du back-office ou de l’outillage. Ça m’a servi de rattrapage sur ces technologies :

HTML/CSS ( ne rigolez pas j’ai appris des choses…)
Bootstrap
Angular
Applications mobiles hybrides
NodeJS

Sachant que j’ai principalement travaillé sur JAVA, ça m’a fait un peu bizarre

Qu’est-ce que j’en ai retenu:

Javascript et les langages interprétés, ce n’est pas fait pour moi. Un compilateur c’est quand même bien pour découvrir des erreurs de syntaxes (je suis trop vieux pour ces co%%%£$$) …. Les puristes du javascript me diront qu’il y a les tests unitaires. Pour moi ça ne remplace pas un compilateur. Pas étonnant que des initiatives comme typescript ou babeljs arrivent massivement sur le marché.

Je comprends pourquoi AngularJS a inondé le marché ces dernières années. C’est assez bien foutu.

Le combo NodeJS/express/mongoose permet de réaliser des APIS et des requêtes facilement .

Un exemple de route pour une requête GET

myRouter.route('/')
    .get(function (req, res, next) {
        model.find({})
                 .exec(function (err, entity) {
                if (err) throw err;
                res.json(entity);
            });
    })

myRouter.route('/')

.get(function (req, res, next) {

model.find({})

.exec(function (err, entity) {

if (err) throw err;

res.json(entity);

});

})

Un exemple de requête qui m’a bluffé

 Model.remove({
            'att1': value
        }, function (err, resp) {
            if (err) throw err;
            res.json(resp);
        });

Model.remove({

'att1': value

}, function (err, resp) {

if (err) throw err;

res.json(resp);

});

Pour conclure, je ne pense pas que je ferai des sites web de sitôt, mais ça m’a donné une bonne vision d’ensemble des technologies.

Devoxx 2016

22 avril 2016 littlewing Laisser un commentaire

Ben voila, j’écris ce billet entre deux confs lors de l’édition 2016 du devoxx.
QUOI, vous ne connaissez pas devoxx ? c’est à mon avis LA conférence française du développement JAVA, SCALA,BIG DATA.

C’est très très technique et c’est tant mieux. Commerciaux et chefs de projets, vous pouvez passer votre chemin 🙂

J’ai pu notamment faire des hands on sur apache kafka, vert.x et sur la création d’architectures.

Petite nouveauté, les conférences seront disponibles sur youtube ( je ne connais pas encore la chaîne ).

Si vous ne connaissez pas, voici quelques conférences de l’année dernière

Quid des données personnelles ?

31 mars 2016 littlewing Laisser un commentaire

Ces derniers mois j’ai pu réaliser deux salons assez intéressants : le BIG DATA PARIS et l’IOT WORLD.
Dans la première conférence, j’ai pu voir pas mal de conférences intéressantes sur les tendances actuelles :

streaming avec Apache Spark
kafka
utilisation du machine learning
…

ainsi que les use cases :

Maintenance prédictive : Grâce au machine learning on arrive à prévoir les pannes et intervenir avant que celle-ci se déclenche. La prochaine étape sera sans doute Minority Report….
Stockage massif des données non structurées ( IOT, logs,…) pour utilisation (marketing) et revente.

Si on exclut les éditeurs qui fournissent des solutions « clés en main ». Toutes les briques de base sont open-source. Beau pied de nez aux éditeurs qui fustigeaient les logiciels libres il y a une dizaine d’années arguant que ce modèle n’est pas viable :).

Dans la deuxième conférence, j’ai vu voir la plateforme AWS IOT. Ça claque, Amazon vend de manière très agressive une solution tout en un, de l’objet avec des kits estampillés « AWS IOT » jusqu’à la plateforme BIGDATA permettant de traiter et d’analyser ces données.

Enfin j’ai pu avoir une perspective sur le futur du marché automobile. L’UE souhaite que les véhicules deviennent connectés à partir de 2019. L’arrivée des GAFA accélère fortement le marché. Les constructeurs automobiles souhaitent changer le business model (achat –> location) et se transformer en fournisseur de données grâce à de nombreux capteurs, une puce 3G obligatoire sur chaque véhicule et près de 150To / véhicule / an. De quoi faire le business des marchands de datacenter et des équipes marketing pendant quelques années.

Outre les points positifs purement techniques que j’ai vu lors de ces conférences, j’ai pu malheureusement noter les points suivants :

L’utilisation massive du Cloud ne garantit en rien la sécurité des données. En effet, près de 30% des données sont revendus à des tiers même si on a coché la bonne case lors de la souscription du contrat.

Un exemple Amazon AWS est soumis au Patriot / Liberty Act . Je vous laisse deviner ce qui peut arriver si Airbus lâche sur le cloud Amazon les plans de ses futurs avions sur le cloud Amazon.

Aussi, les dispositifs législatifs actuels sont sous dimensionnés pour lutter contre la réutilisation frauduleuse des données. La CNIL évolue ( les amendes vont passer de 170 K€ à 4% du CA), mais les avocats qui étaient aux conférences indiquaient que passer des contrats avec des sociétés étrangères à l’UE était un risque

Enfin, on va réussir grâce au BIG DATA à corréler d’innombrables données qui permettront dans le meilleur des cas de faire pas mal d’optimisations (ex. la ville de new york), mais aussi de connaître le comportement des gens et même de le prévoir.

Tout ça pour dire qu’on va être de plus en plus fliqué. Je m’en doutais, mais c’est dorénavant pour moi une certitude. Le « BIG BROTHER » nous attend désormais au coin de la rue. Que ça soit par les états ou pire par les sociétés. Pour illustrer mon propos, et je finirai là, je pourrais prendre pour exemple les voitures connectées. Les intervenants représentant les constructeurs indiquaient que les données étaient soit-disant anonymisées mais que les forces de l’ordre pouvaient y avoir accès et qu’il pouvaient réaliser des conseils personnalisés…. Bref, ce n’est pas anonymisé pour tout le monde 🙁

Intégrer dans docker une application ANGULAR

21 mars 2016 littlewing Laisser un commentaire

Dans le cadre d’un POC j’ai eu à « dockeriser » une application ANGULARJS. Cette dernière utilise les briques logicielles suivantes :

Angular JS
Json-server qui se abse sur nodeJS pour bouchonner mes APIS REST.

J’ai donc utilisé docker-compose pour orchestrer le tout. Voici la configuration

front:
  build: ./front
  ports: 
    - "80:80"
back:
  image: clue/json-server
  ports:
    - "3000:80"
  volumes:
    -  $PWD/back/config:/data

front:

build: ./front

ports:

- "80:80"

back:

image: clue/json-server

ports:

- "3000:80"

volumes:

- $PWD/back/config:/data

Définition du front

Dans le répertoire front, j’ai ajouté les sources (JAVASCRIPT, HTML, CSS,…). Ca donne

FROM coreos/apache
MAINTAINER ....
EXPOSE 80
COPY resources/monsite /var/www

CMD ["/usr/sbin/apache2ctl","-D","FOREGROUND"]

FROM coreos/apache

MAINTAINER ....

EXPOSE 80

COPY resources/monsite /var/www

CMD ["/usr/sbin/apache2ctl","-D","FOREGROUND"]

Définition du back

Pour le back, vu que c’est du mock, je ne me suis pas trop embêté. J’ai donc utilisé json-server et ajouté dans le répertoire config le fichier json correspondant aux bouchons.

Exécution

Construction du projet

A la racine

$ docker-compose build

1	$ docker-compose build

Exécution

$docker-compose up

1	$docker-compose up

Conclusion

Et voila vous avez un site basé sur angular avec utilisation d’une API REST utilisable . Il manque encore certaines briques indispensables telles que ha_proxy par ex, mais ça me suffit pour une démo.

Développement WEB

10 mars 201610 mars 2016 littlewing Laisser un commentaire

Bon je n’ai pas écrit d’article depuis quelques temps. Je suis à fond sur l’apprentissage des technologies WEB :

bootstrap
angular
yeoman
nodejs
…

J’ai donc un peu laché le développement spark/scala pour le parcours de spécialisation coursera.

Je pense écrire un peu plus ensuite ( dans quelques mois …)

See you

Paramétrage de ma Debian effectué post-installation

14 janvier 2016 littlewing Laisser un commentaire

après avoir réinstallé mon PC et optimisé les accès au disque SSD, me voici dans le téléchargement des packages

Sources logicielles

J’ai copié mes anciennes sources de packages dans le répertoire /etc/apt/sources.list.d.

J’ai exclu la source deb-multimedia.org.

Packages installés

Attention, certains packages sont spécifiques à ma plateforme! A vous de voir si vous voulez les utiliser

#apt-get install gnome-disk-utility virtualbox-5.0 ttf-mscorefonts-installer \
 nvidia-kernel-dkms pavucontrol alsamixergui mudita24 w32codecs gstreamer1.0-plugins-bad \
easytag tuxguitar-jsa lilypond htop geany frescobaldi filezilla gparted grsync \
openjdk-8-jdk chromium  flashplugin-non-free autofs

#apt-get install gnome-disk-utility virtualbox-5.0 ttf-mscorefonts-installer \

nvidia-kernel-dkms pavucontrol alsamixergui mudita24 w32codecs gstreamer1.0-plugins-bad \

easytag tuxguitar-jsa lilypond htop geany frescobaldi filezilla gparted grsync \

openjdk-8-jdk chromium flashplugin-non-free autofs

Configuration nvidia

Il faut créer le fichier /etc/X11/xorg.conf.d/20-nvidia.conf avec le contenu suivant :

Section "Device"
 Identifier "NVIDIAGeForce210"
 Driver "nvidia"
 EndSection

Section "Device"

Identifier "NVIDIAGeForce210"

Driver "nvidia"

EndSection

API utilisées

Configuration maven

Développement

MySourceConnector

MySourceTask

Configuration nécessaire à l’exécution du plugin

Exécution

Conclusion

Installation

Configuration du connecteur JDBC pour KAFKA CONNECT

Exécution

Conclusion

Rappels des épisodes précédents

Pré-requis

Configuration Logstash

Mapping ELASTICSEARCH

Analyzers

Mapping

Création de l’index

Chargement

Interrogation

Conclusion

Définition du front

Définition du back

Exécution

Construction du projet

Exécution

Conclusion

Sources logicielles

Packages installés

Configuration nvidia

Catégories

Méta

Étiquettes