Verteilte Anwendung komplexer Filteroperationen auf große Bilder

(This is in German. I will translate it, if enough comments asking me to do it ;-))

Bei der Aufklärung werden zur Auswertung von Satellitenbildern derzeit komplexe Filteroperationen auf dem jeweiligen Auswerter-PC ausgeführt. Die Filter können der allgemeinen Bildverbesserung dienen oder auch dafür sorgen, dass bestimmte Objekte wie Strassen oder Fahrzeuge auf einem Satellitenbild besser und schneller zu erkennen sind. Wenn die Bilder eine gewisse Größe von mehreren Gigabyte erreichen, kann die dazu notwendige Rechenzeit schon unangenehm werden. Besonders dann, wenn mehrere solcher Filter nacheinander ausgeführt werden müssen.

Bisher geschah dies zumeist auf der Client Seite der Auswerte-PC. Obwohl Arbeitsplatzrechner immer schneller werden, ist ein automatisches, serverseitiges Batchprozessing sinnvoll, um einerseits die Filter Berechnungen nur einmal ausführen zu müssen und zum anderen, um die notwendige Rechenleistung auf mehrere Server-Maschinen verteilen zu können.

Aber nicht nur das Berechnen der Bilder ist ein Problem, auch die Datenmengen, die gespeichert und über die Netzwerke übertragen werden müssen. Bei einem Netz von über 50 Auswerte-PCs können nicht ständig die vollständigen Bilder übertragen werden. Die Bilddateien müssen in kleinere Stücke zerlegt werden, sogenannte Kacheln. Bei der Anforderungen eines bestimmten Ausschnitte zur Ansicht, muss im einfachsten Fall nur eine Kachel anhand ihrer Koordinaten gefunden und übertragen werden. Desweiteren können die einzelnen Kacheln noch in unterschiedlichen Auflösungen gespeichert und je nach angezeigter Zoom-Stufe übertragen werden.

Hadoop

Betrachtet man die Anforderungen, lassen sich die notwendigen Schritte recht einfach beschreiben.

  • Zerlegen des Bildes in Kacheln
  • auf mehrere Maschinen verteiltes Ausführen von Filteroperationen
  • Ablage der Bilddaten in eine Datenbank
  • Berechnungen eines spatial Indexes, zum schnellen Finden der zum angeforderten Bildausschnitt passenden Kacheln

Hadoop bietet zum einen die Möglichkeit eingehende Datensätze zu zerteilen (splitten) und durch seine MapReduce Konzepte verteilt zu bearbeiten, zum anderen die NoSQL Datenbank HBase um binäre Bildaten persistieren zu können.
Da Lucene einen spatial Index beherrscht, ist dies das Tool der Wahl, um die Bilder in der Datenbank auch wieder zu finden.

Kacheln

Das Zerlegen des Bildes in Kacheln ist nur auf den ersten Blick einfach, schließlich soll die Georeferenzierung des Gesamtbildes korrekt auf die einzelnen Kacheln übertragen und sich die Filegrösse auch ungefähr der Blockgrösse des HDFS anpassen. Hier beschränken wir uns auf einen quasi Standard für solches Bildmaterial dem GeoTIFF Format. Es bietet zusätzlich zum TIFF eine Erweiterung zur Speicherung von Daten zur Georeferenzierung. Verwendet wird dazu eine GeoTIFF library für Java Advanced Imaging JAI. Sie bietet bereits die Möglichkeit aus einem grossen Bild einzelne Kacheln zu erzeugen, wie auch die automatische Beibehaltung der Geoinformationen.
Nun muss nur noch eine Möglichkeit gefunden werden, die einzelnen Kacheln den Map Prozessen zuzuweisen. Die einfachste ist das SequenceFile Format. Im Prinzip ein Key-Value-Store von beliebigen Binärarrays in einem einzigen File.

Wir wandeln also das Bild in einzelne Kacheln und speichern die Binärdaten in ein Sequence File. Dabei werden die Kacheln mit einer Nummer versehen, die als Key verwendet wird (siehe Bild 1).
Der folgende Code erstellt ein solches SequenceFile an einem gegebenen Pfad auf Hadoops verteiltem Dateisystem und fügt die einzelnen Key/Value Paare per append (+=) Funktion an. Dabei wird als Key ein simpler String (Text Objekt) und als Value ein Byte-Array (BytesWritable Objekt) verwendet.

  def aquire {
    doAppend[Text, BytesWritable]("hdfs://...", "ImageSequenceFile") {
      (sf, k, v) =>
        for { ... } {
          val imageArray = tile.getRGB(. . .)
          val key = ....
          k.set(key)
          v.set(imageArray, 0, imageArray.length)
          sf += (k, v)
        }
    }
  }

Damit ist die Konvertierung eines Bildes in Kacheln und die Aufbereitung der Daten für nachfolgende Map-Jobs schon erledigt.

MapReduce

Map und Reduce gehören in normalerweise eigentlich zusammen. Allerdings wollen wir in unserem Fall das Bild nicht mehr zu einem Gesamten zusammen fassen, sondern die Kacheln getrennt in HBase speichern. Ein klassischer Reduce-Job kann also entfallen.

Die Bilder liegen bereits in einem SequenceFile vor. Das obige Bild zeigt, wie jeder Map-Job jeweils ein Key/Value paar zu Bearbeitung zugewiesen bekommt. Da die Bilder als Value abgelegt sind, können wir die Filteroperationen darauf anwenden. Die Ergebnisse werden in jeweils eine Zeile der HBase geschrieben. Dazu wird

  • Als RowKey den Name des ursprünglichen Bildes und ein Nummernpaar für die Kachel verwendet
  • Für Metainformationen folgende Column Families verwendet
    • “loc”, um die Positionsinformationen und
    • “info”, um allgemeine weitere Informationen über die Kachel, wie Bildherkunft etc. zu speichern.
  • Das Bild selbst in unterschiedlichen Auflösungen und Berechnungen in der Column Family “images” gesichert.

Die HBase “Tabelle” wurde durch folgendes Statement in der Shell erzeugt:

create 'images', {NAME => 'image', VERSIONS => 1}, {NAME => 'info', VERSIONS => 1}, {NAME => 'loc', VERSIONS => 1}, {NAME => 'loc', VERSIONS => 1}

Der Start der Map Jobs und die Zuweisung der einzelnen Keys erledigt das Hadoop Framework automatisch. Das heisst, dass die verfügbaren Knoten je einen Key zur Bearbeitung zugewiesen bekommen.

object LargeImageProcessingJob {
  def main(args: Array[String]):Unit = {
    val job = new Job(...)
    FileInputFormat.setInputPaths(job, "hdfs://.../ImageSequenceFile")
    job.setJobName("ImageConverter")
    job.setInputFormatClass(classOf[SequenceFileInputFormat[_,_]])
    job.setMapperClass(classOf[GrayScaleMapper])
    job.setNumReduceTasks(0)
  }
}

Durch den gezeigten Code wird der Input Pfad gesetzt (auf unser oben erzeugtes SequenceFile), die Anzahl der Reduce-Jobs auf 0 gesetzt und das Klassenobjekt für den Map-Job festgelegt.

Der eigentliche Code zum Map-Job sieht folgender maßen aus:

class GrayScaleMapper extends Mapper[Text, BytesWritable, Text, BytesWritable] with HBaseWrapper {

  val conf = HBaseConfiguration.create
  val hTable = new HTable(conf, "images")

  override def map(key: Text, file: BytesWritable,
          context: Mapper[Text, BytesWritable, Text, BytesWritable]#Context) = {

    val split = context.getInputSplit.asInstanceOf[FileSplit]

    // convert byte array to BufferedImage
    import Conversions._
    val intArray:Array[Int] = file.getBytes
    println("available Ints: + " + intArray.size)

    // get size of image from Key text
    val ds = key.toString.split(" ")
    val (width, height) = (ds(ds.size-2).toInt, ds(ds.size-1).toInt)
    println("Image with Width: " + width + " and Hight: " + height)

    // create BufferedImage from Integer Array
    val buffImg = new BufferedImage (width, height, BufferedImage.TYPE_BYTE_GRAY)
    buffImg.getRaster.setPixel(0, 0, intArray)

    // grayscale the image's pixel
    val op = new ColorConvertOp(ColorSpace.getInstance(ColorSpace.CS_GRAY), null)
    val grayImage = op.filter(buffImg, null)

    // convert BufferedImage to byte array
    val baos = new ByteArrayOutputStream

    ImageIO.write(grayImage, "tif", baos)
    val imageAsByteArray = baos.toByteArray

    println("putting image with key: " + key.toString)
    doPut(hTable, key.toString) {
      p =>
        p += ("image", "standard", imageAsByteArray)

        p += ("info", "name", key.toString);
        p += ("info", "description", "Image Description from Image Matadata")

        p += ("loc", "lat", "Location... Latitude")
        p += ("loc", "lon", "Location... Longitude")
        p += ("loc", "extend", "Zoom... Extend")
    }
  }
}

Der Mapper holt die Rawbytes des Bildes aus seinem übergebenen split. Daraus wird dann wieder ein BufferedImage zur weiteren Bearbeitung erzeugt. Hier muss durch komplexe Bildfilter der größte Teil der Rechenzeit anfallen. So könnten z.B. neben einer Bildveränderung, unterschiedliche Auflösungen berechnet werden. Das hier gezeigt Verfahren ist je besser desto mehr an Filteroperationen hier notwendig werden.
Im gezeigten Code wird eine einfache Konvertierung in ein Grauwertbild verwendet. Ist die Berechnung abgeschlossen wir über die ImageIO ein TIF Bild erzeugt. Dessen Bytearray wird direkt zum beschreiben der Column Family “image” verwendet. Sollten mehrere Auflösungen berechnet worden sein, könnte man neben dem Key “standard” beispielsweise “-8” oder “-16” usw. verwenden.

Lucene

Die Bilder sind gespeichert, allerdings müssen sie auch wieder gefunden werden. Da die Bilder georeferenziert sind ist eine Aufgabe anhand des, auf einem Client, gezeigten Ausschnitts über die Koordinaten die notwendigen Bilder zu finden. Dazu verwenden wir Lucene und den zur Verfügung gestellten Spatial-Index. Allerdings gibt es ein Problem: Lucene kann nur sogenannte Distance Queries d. .h., es können nur die Punkte gefunden werden, die sich innerhalb eines Kreises befinden. Da der anzuzeigender Bildausschnitt ein Rechteck ist, bleibt nichts anderes als das folgende übrig:

Der Einfachheit halber kann man davon ausgehen, dass alle Bilder eine rechteckige Form haben (tatsächlich sind projizierte Bilder nie genau rechteckig. Bei einigen Sensorbildern können sogar noch nicht einmal die begrenzenden Koordinaten festgelegt werden, da wegen großer Aufnahmewinkel von Horizont zu Horizont fotografiert wird).
Wir geben jeder Ecke (in der Abbildung als rote Pfeile symbolisiert) eine latitude und longitude Koordinate, also einen Punkt. Alle 4 Punkte werden mit dem selben HBase-RowKey des Bildes durch Lucene indexiert, zeigen also alle auf das selbe Bild. Den “Such”-Umkreis berechnen wir aus dem zu zeigenden Bildausschnitt (in der Abbildung gelb). Alle 4 Ecken des Bildausschnitts befinden sich auf dem Umkreis.
Zusammen mit dem Kreis Mittelpunkt lässt sich die Lucene Distance Query ausführen.

. . .
val dq = new DistanceQueryBuilder(
      lat_Umkreis,
      lng_Umkreis,
      radius,
      latField,
      lngField,
      tierPrefix,
      true)
. . .

Zurück erhalten wir eine Liste mit Bildpunkten, sie sich innerhalb des Kreises befinden. Über Lucenes hits = searcher.search(. . .) und die hits.scoreDocs können wir eine Liste der Bilder erhalten.
Dies ist allerdings immer (nur) mindestens die Anzahl Bilder, die auch angezeigt werden müssen. Aber ist ist zumindes eine Art Vorauswahl. Sollte dies nicht genügen, kann über genauere Geometrien und Intersections davon die Auswahl weiter präzisiert werden. Dazu könnte z. B. die Java Topologie Suite# verwendet werden.

Fazit

Um nicht mit Kanonen auf Spatzen zu Schiessen, müssen bei diesem Ansatz drei Bedingungen erfüllt sein

  • die Bilder müssen eine signifikante Größe haben
  • die Filteroperationen müssen komplex genug sein um sinnvoll verteilt werden zu können
  • der Zugriff auf wechselnde Bilddateien muss schnell über Koordinaten erfolgen können

Ist dies alles der Fall und benötigt man eine Art vorprozessierte und normalisierte Ablage der Bilddateien zum schnellen Zugriff sind die Möglichkeiten von MapReduce und das verteilte Dateisystem HFS ideal. Reicht der zur Verfügung stehende Plattenplatz nicht aus, kann leicht weitere billige Hardware hinzugefügt werden. Oder umgekehrt, muss schneller bearbeitet werden wird durch neue Hardware die Rechenleistung erhöht und nebenbei der Speicherplatz ebenfalls.
Natürlich muss man bedenken, dass MapReduce und das erstellen oder neu erstellen des Spatial Indexes auch Zeit kostet. Bei wenigen, kleinen Bildern macht dies wenig Sinn. Aber umso mehr je grösser die Bilder sind, je komplexer die Filteroprationen und je schneller der Zugriff sein muss.

PS

Der hier dargestellte Code ist unvollständig und soll die Funktionalität nur exemplarisch zeigen.
Eine sehr frühe (nicht funktionierende) Version der Quellen liegt auf Github. Ich werde versuchen die Endversion des Proof of Concepts von Codeteilen zu bereinigen, die ich nicht veröffentlichen darf und damit die Github Version zu ersetzen. Was allerdings etwas dauern wird… Ich hoffe aber, dass der Artikel trotzdem interessant ist.

Connect with me on Google+

2 thoughts on “Verteilte Anwendung komplexer Filteroperationen auf große Bilder

  1. Dieser Ansatz ist gut für Filter die über einzelne Pixel wirken – was ist mit Filtern die auf einen Region basieren ( Median / Gauss / Sauvola etc. ) – da muss man Überlappung zulassen

  2. Die Kacheln müssen einfach überlappend, also etwas grösser, ausgeschnitten werden. Nach der Berechnung kann man sie dann wieder zurecht schneiden, wenn man möchte.