2

DryadLINQComment passer à l'échelle le traitement de données volumineuses ?Xavier PillonsPrincipal Program ManagerMicrosoft Corporation

TC301

3

Le marché du « Data Intensive Scalable Computing » (DISC)• Le spectre des besoins des utilisateurs de DISC est

large• A une extrémité se trouve l’analyse traditionnelle des

entrepôts de données structurées. L’analyste connait le cube qu’il/elle désire construire ainsi que les sources de données.

• Une autre extrémité est l’analyse de données brutes non structurées. L’analyste ne connait pas exactement ce que contiennent les données ni quel cube pourrait être construit. L’analyste a besoin de traitements ad-hoc qui ne pourraient jamais être rejoués.

• Windows HCP Server+Dryad cible l’analyse de données brutes non structurées.

4

Windows HPC+Dryad en complément de l’offre Microsoft• Microsoft a déjà une excellente plateforme

de gestion des données• SQLServer Integration Services (SSIS),

PowerPivot, Parallel Data Warehouse (PDW), Stream Insight…

• HCP Server+Dryad cible l’analyse de données brutes non structurées• Permet de nouvelles solutions mixant plusieurs

technologies de traitement des données• Ex : HPC+Dryad en entrée de SSIS puis d’un

traitement BI plus traditionnel

5

Un exemple de mise en œuvre de DISC• Razorfish vend du marketing de publicités

ciblés sur le net.• Besoin d’analyser des TeraOctets de clics sur les

sites de leurs clients pour cibler les visiteurs et leur montrer une annonce appropriée.

• L’analyse est nettement moins cher avec un système DISC qu’avec une super base SQL

• Razorfish a mis en œuvre Hadoop (Implémentation OSS du concept MapReduce)

6

Quelle est la réponse de Microsoft ?

7

Dryad est une technologie utilisée en interne chez Microsoft pour des scénarios DISC• Dryad a été utilisé par Microsoft Search

pendant 5 ans sur des milliers de serveurs• Dryad est à la pointe de la recherche et du

calcul distribué• Dryad est une technologue éprouvée et

nous la rendons disponible via Windows HPC Server 2008 SP1

8

Dryad / DryadLinq / DSC

Déploiement HPC, ordonnancement, etc.

MPI SOA Dryad

Windows Server Azure

Runtime distribué

Services pour Cluster et Cloud

Plateforme

Distributed Storage Catalog

Assemble des partages NTFS pour le moteur distribué de Dryad

Ajouts spécifiques pour le DISC

C#, C++, …C#/LINQ pour Dryad

Visual Studio, Excel, etc.Visual Studio pour

C#/LINQ

Langages et Librairies

Outils de développement

9

Dryad permet au HPC de résoudre de nouveaux types de problèmes

MPI SOA

Dryad Optimise le déplacement des données plutôt que l’usage du CPU afin de

faciliter les travaux intensifs en E/S.

Optimise l’usage du CPU pour les problèmes

faiblement couplés comme la simulation de produits financier complexes, etc.

Optimise l’usage du CPU pour des problèmes

fortement couplés comme la modélisation du climat, la mécanique des fluides,

etc…Orienté CPU

Orienté Données

10

HPC + Dryad adressent le marché du DISC pour les clusters et Azure• L’utilisateur écrit des programmes à l’aide de

Visual Studio. Puis Windows HPC + Dryad exécute ces programmes en les passant à l’échelle.

HPC Scheduler

Visual Studio

Pour l’utilisateur: Exécuter un programme C#En interne : Soumet un travail HPC

HPC Compute

Node

Sur un Cluster: Exécute des calculs qui lisent/écrivent beaucoup de fichiers.

HPC Azure Node

Dans Azure: Exécute des calculs qui lisent/écrivent beaucoup de blobs.

11

‘Pipe’ à 2 Dimensions

Pipe Unix: 1-Dgrep | sed | sort | awk | perl

Dryad: 2-D grep1000 | sed500 | sort1000 | awk500 | perl50

12

Un job Dryad est un graphe acyclique dirigé (DAG)

Vertex d’éxécution

Canauxd’échanges

Entrées

Sorties

13

Un besoin simple

Pages web les plus visitées par xavier

var logentries = from line in logs where !line.StartsWith("#") select new LogEntry(line);var user = from access in logentries where access.user.EndsWith(@"\xavier") select access;var accesses = from access in user group access by access.page into pages select new UserPageCount(“xavier", pages.Key, pages.Count());var htmAccesses = from access in accesses where access.page.EndsWith(".htm") orderby access.count descending select access;

14

Les Etapes de cette requêtevar logentries = from line in logs where !line.StartsWith("#") select new LogEntry(line);var user = from access in logentries where access.user.EndsWith(@"\xavier") select access;var accesses = from access in user group access by access.page into pages select new UserPageCount(“xavier", pages.Key, pages.Count());var htmAccesses = from access in accesses where access.page.EndsWith(".htm") orderby access.count descending select access;

Parcours logs et garde seulement les lignes qui ne sont pas des commentaires. Transforme chaque ligne en un objet LogEntry.

Parcours logentries et conserve seulement les entrées correspondant à xavier.

Groupe les accès de xavier en fonction de chaque page correspondante. Pour chaque page compte le nombre d’occurence.

Trie les pages accédées par xavier en fonction de leur fréquence d’accès.

15

Exécution séquentiellevar logentries = from line in logs where !line.StartsWith("#") select new LogEntry(line);var user = from access in logentries where access.user.EndsWith(@"\xavier") select access;var accesses = from access in user group access by access.page into pages select new UserPageCount(“xavier", pages.Key, pages.Count());var htmAccesses = from access in accesses where access.page.EndsWith(".htm") orderby access.count descending select access;

Pour chaque ligne de logs faire …

Pour chaque entrée dans logentries faire …

Trie les entrées dans user par page. Puis itèrer sur la liste triée et compter les occurences de chaque page.

Trier à nouveau les entrées de access par fréquence d’accès aux pages.

16

Exécution parallèle avec Dryad

var logentries = from line in logs where !line.StartsWith("#") select new LogEntry(line);var user = from access in logentries where access.user.EndsWith(@"\xavier") select access;var accesses = from access in user group access by access.page into pages select new UserPageCount(“xavier", pages.Key, pages.Count());var htmAccesses = from access in accesses where access.page.EndsWith(".htm") orderby access.count descending select access;

logentries

user

logs

accesses

htmAccesses

output

17

Vue d’ensemble d’un job HPC + Dryad

Application qui utilise les APIs de Dryad

HPC Head Node

DSC

Soumet un Job Dryad

1

1

Le job Dryad démarre aussi un ensemble de tâches « parametric

sweep » sur les autres nœuds en tant que DVH

2b

Un Job Dryad démarre une tâche simple en

assignant à un nœud le rôle de DGM

2a

2a

Les vertex Dryad lisent et écrivent les fichiers de données

3b

Dryad Graph Manager démarre/arrête les

vertex Dryad

3a

HPC Compute Nodes

3a

3b2b

Dryad Graph Manager

Dryad Vertex Host

18

Dryad exécute le graphe en associant des vertex aux instances des Dryad Vertex Hosts

Vertex d’éxécution

Canauxd’échanges

Entrées

Sorties

Ressources disponibles

19

Mécanismes internes

Application qui utilise les APIs de Dryad

HPC Head Node

DSC1

Le job Dryad démarre aussi un ensemble de tâches « parametric

sweep » sur les autres nœuds en tant que DVH

2b

Un Job Dryad démarre une tâche simple en

assignant à un nœud le rôle de DGM

2a

2a

HPC Compute Nodes

3a

3b2b

Dryad Graph Manager

Dryad Vertex Host

Publie sur un partage:1. Binaires pour le job Dryad

2. Description XML du graphe Dryad

1

Le DGM charge la description XML du graphe Dryad et contact le DSC pour localiser les fichiers de

données

3a

Les DVH chargent les binaires de ce job Dryad depuis le partage

réseau et les exécute en fonction des commandes issues du DGM

3b

using System;using System.Linq;using System.IO;using System.Security.Cryptography;

namespace DupPic1{ class Program { static void Main(string[] args) { string directoryName = @"C:\Temp\pics"; var duplicatedFiles = Directory.GetFiles(directoryName,"*.jpg", SearchOption.AllDirectories) .Select(filename => new { hash = GetChecksum(filename), name = filename }) .GroupBy(record => record.hash) .Where(group => group.Count() > 1) .SelectMany(group => group.Select(record => record.name));

foreach (var file in duplicatedFiles) { Console.WriteLine(file); } }

static string GetChecksum(string file) { using (FileStream stream = File.OpenRead(file)) { SHA256Managed sha = new SHA256Managed(); byte[] checksum = sha.ComputeHash(stream); return BitConverter.ToString(checksum).Replace("-", String.Empty); } } }}

Pour chaque fichier calculer le CheckSum

Grouper par checksum identiques

Conserver seulement les groupes dont le nombre est >1

Sélectionner le nom des fichiers doublons

Lister tous les fichiers *.jpg et créer un IEnumerable<string>

using System;using System.Linq;using System.IO;using System.Security.Cryptography;using Microsoft.Distributed.Linq;

namespace DupPic2{ public class Program { static void Main(string[] args) { string directoryName = @"\\mymachine\samples\pics"; var duplicatedFiles = Directory.GetFiles(directoryName, "*.jpg", SearchOption.AllDirectories) .AsDistributed() .Select(filename => new { hash = GetChecksum(filename), name = filename }) .GroupBy(record => record.hash) .Where(group => group.Count() > 1) .SelectMany(group => group.Select(record => record.name));

foreach (var file in duplicatedFiles) { Console.WriteLine(file); } }

public static string GetChecksum(string file) { using (FileStream stream = File.OpenRead(file)) { SHA256Managed sha = new SHA256Managed(); byte[] checksum = sha.ComputeHash(stream); return BitConverter.ToString(checksum).Replace("-", String.Empty); } } }}

La seule différence dans le code pour une exécution distribuée est .AsDistributed()

Marquer la classe comme publique

Marquer la méthode comme publique

Espace partagé de stockage des images

Ajouter en référence Microsoft.Distributed.Linq

namespace DupPic3{ public class Program { static void Main(string[] args) { string directoryName = @"\\mymachine\samples\pics";

Directory.GetFiles(directoryName, "*.jpg", SearchOption.AllDirectories) .AsDistributed() .Execute("hpcdsc://myheadnode/samples/images");

var duplicatedFilesQuery = DistributedData<string> .Open("hpcdsc://myheadnode/samples/images") .Select(filename => new { hash = GetChecksum(filename), name = filename }) .GroupBy(record => record.hash) .Where(group => group.Count() > 1) .SelectMany(group => group.Select(record => record.name));

var duplicatedFiles = duplicatedFilesQuery.Execute("hpcdsc://myheadnode/samples/duplicatedImages"); foreach (var file in duplicatedFiles) { Console.WriteLine(file); } }

public static string GetChecksum(string file) { using (FileStream stream = File.OpenRead(file)) { SHA256Managed sha = new SHA256Managed(); byte[] checksum = sha.ComputeHash(stream); return BitConverter.ToString(checksum).Replace("-", String.Empty); } } }}

Création dans le DSC d’un flux identifiant les fichiers

Ouverture du flux précédemment créé

Exécution du graphe Dryad et écriture du résultat dans un nouveau flux du DSC

23

En Conclusion• Windows HPC+Dryad est la solution Microsoft aux

solution DISC dans un cluster et/ou dans Azure

• L’avantage du DISC Microsoft réside dans son offre de bout en bout et pas sur Dryad uniquement

• Utiliser Linq pour le traitement de vos données• L’utilisation de Dryad n’en sera que plus rapide

• Essayer le et dites nous ce que vous en pensez• http://connect.microsoft.com

24

Sessions relatives• MongoDB et C#: quand .NET rencontre NoSQL

(INT201) – Mardi 17h30• Nouveautés Windows HPC 2008 R2 et le vol de

cycles (TC201) – Mercredi 11h00• HPC Services pour Excel 2010 (TC302) – Mercredi

13h00

25

Références• Windows HPC Server 2008 SP1

• http://www.microsoft.com/hpc• Dryad / DryadLinq / DSC

• http://connect.microsoft.com• Language-Integrated Query (LINQ)

• http://msdn.microsoft.com/en-us/library/bb397926.aspx

26

MSDN et TechNet : l’essentiel des ressources techniques à portée de clic

http://technet.com http://msdn.com

Portail administration et infrastructure pour informaticiens

Portail de ressources technique pour développeurs