Monte-Carlo-Methoden mit Dataproc und Apache Spark

Dataproc-Cluster erstellen

Führen Sie die Schritte zum Erstellen eines Dataproc-Clusters über die Google Cloud Console aus. Die Standardclustereinstellungen, die zwei Worker-Knoten umfassen, sollten für diese Anleitung ausreichend sein.

Logging für Warnungen deaktivieren

Standardmäßig druckt Apache Spark ausführliche Aufzeichnungen im Konsolenfenster. Für ein besseres Verständnis dieser Anleitung sollten Sie die Logging-Ebene so ändern, dass nur Fehler in Logs aufgezeichnet werden. Gehen Sie so vor:

ssh-Verbindung zum primären Knoten des Dataproc-Clusters herstellen

Der primäre Knoten des Dataproc-Clusters hat das Suffix -m im VM-Namen.

1. In the Google Cloud console, go to the VM instances page.

   Go to VM instances

2. In the list of virtual machine instances, click SSH in the row of the instance that you want to connect to.

   

Ein SSH-Fenster wird geöffnet, das mit dem primären Knoten verbunden ist.

Connected, host fingerprint: ssh-rsa 2048 ...
...
user@clusterName-m:~$

Logging-Einstellungen ändern

1. Bearbeiten Sie im Stammverzeichnis des primären Knotens /etc/spark/conf/log4j.properties.

   sudo nano /etc/spark/conf/log4j.properties
   

2. Setzen Sie log4j.rootCategory auf ERROR.

   # Set only errors to be logged to the console
   log4j.rootCategory=ERROR, console
   log4j.appender.console=org.apache.log4j.ConsoleAppender
   log4j.appender.console.target=System.err
   log4j.appender.console.layout=org.apache.log4j.PatternLayout
   log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
   

3. Sichern Sie die Änderungen und beenden Sie den Editor. Wenn Sie das ausführliche Logging wieder aktivieren möchten, können Sie die Änderung rückgängig machen. Setzen Sie dazu den Wert für .rootCategory auf den ursprünglichen Wert (INFO) zurück.

Spark-Programmiersprachen

Spark unterstützt Python, Scala und Java als Programmiersprachen für eigenständige Anwendungen und bietet interaktive Interpretierer für Python und Scala. Es ist Ihnen überlassen, welche Sprache Sie verwenden. In dieser Anleitung werden interaktive Interpretierer genutzt, weil Sie damit experimentieren können. Zu diesem Zweck können Sie Code ändern, verschiedene Eingabewerte testen und dann die Ergebnisse aufrufen.

Portfoliowachstum schätzen

Im Finanzwesen werden Monte-Carlo-Methoden manchmal verwendet, um Simulationen durchzuführen, die versuchen, das mögliche Ergebnis einer Investition vorherzusagen. Indem Stichproben von Ergebnissen für eine Reihe von wahrscheinlichen Marktbedingungen erstellt werden, kann eine Monte-Carlo-Simulation Fragen dazu beantworten, wie das Ergebnis eines Portfolios im Durchschnitt oder im schlimmsten Fall aussehen könnte.

Befolgen Sie diese Schritte zum Erstellen einer Simulation, die Monte-Carlo-Methoden verwendet, um zu versuchen, das Wachstum einer finanziellen Investition auf der Grundlage einiger üblicher Marktfaktoren abzuschätzen.

1. Starten Sie den Python-Interpretierer über den primären Dataproc-Knoten:

   pyspark
   

   Warten Sie auf die Spark-Eingabeaufforderung >>>.

2. Geben Sie den folgenden Code ein. Achten Sie darauf, dass die Einrückung in der Funktionsdefinition beibehalten bleibt:

   import random
   import time
   from operator import add
   
   def grow(seed):
       random.seed(seed)
       portfolio_value = INVESTMENT_INIT
       for i in range(TERM):
           growth = random.normalvariate(MKT_AVG_RETURN, MKT_STD_DEV)
           portfolio_value += portfolio_value * growth + INVESTMENT_ANN
       return portfolio_value
   

3. Drücken Sie return, bis Sie die Spark-Eingabeaufforderung wieder sehen.

   Der vorhergehende Code definiert eine Funktion, die zeigt, was passieren könnte, wenn ein Investor ein bestehendes Ruhestandskonto hat, das in den Aktienmarkt investiert wird und dem jährlich Geld hinzugefügt wird. Die Funktion generiert für einen bestimmtem Zeitraum jährlich eine zufällige prozentuale Rendite der Investition. Sie nimmt einen Seed-Wert als Parameter. Dieser Wert wird verwendet, um den Zufallsgenerator zurückzusetzen, damit die Funktion nicht jedes Mal die gleiche Liste an Zufallszahlen erhält. Die Funktion random.normalvariate sorgt dafür, dass für die angegebene mittlere und Standardabweichung Zufallswerte über eine Normalverteilung auftreten. Die Funktion erhöht den Wert des Portfolios um den Wachstumsbetrag, der positiv oder negativ sein kann, und fügt eine jährliche Summe hinzu, die weitere Investitionen darstellt.

   Sie legen die erforderlichen Konstanten in einem der nächsten Schritte fest.

4. Erstellen Sie zur Versorgung der Funktion zahlreiche Werte. Geben Sie bei der Spark-Eingabeaufforderung den folgenden Code ein, der 10.000 Werte erzeugt:

   seeds = sc.parallelize([time.time() + i for i in range(10000)])
   

   Das Ergebnis des Vorgangs parallelize ist ein Resilient Distributed Dataset (RDD), d. h. eine Sammlung von Elementen, die für die Parallelverarbeitung optimiert wurde. In diesem Fall enthält der RDD Werte, die auf der aktuellen Systemzeit basieren.

   Bei der Erstellung des RDD segmentiert Spark die Daten basierend auf der Anzahl der verfügbaren Worker und Kerne. In diesem Fall wählt Spark acht Segmente, ein Segment für jeden Kern. Das ist für diese Simulation, welche 10.000 Daten hat, in Ordnung. Bei größeren Simulationen kann jedes Segment größer sein als die Standardgrenze. In diesem Fall kann die Angabe eines zweiten Parameters zum Parallelisieren mit parallelize die Anzahl der Segmente erhöhen, was dazu beitragen kann, die Größe der Segmente überschaubar zu halten, während Spark weiterhin alle acht Kerne nutzt.

5. Fügen Sie der Wachstumsfunktion das RDD hinzu:

   results = seeds.map(grow)
   

   Die Methode map übergibt jeden Wert im RDD an die Funktion grow und hängt jedes Ergebnis an einen neuen RDD an, der in results gespeichert ist. Achten Sie darauf, dass dieser Vorgang, der eine Transformation durchführt, nicht sofort Ergebnisse produziert. Spark wird dies erst tun, wenn die Ergebnisse benötigt werden. Diese verzögerte Bewertung ist der Grund dafür, dass Sie den Code eingeben können, ohne die Konstanten festzulegen.

6. Geben Sie einige Werte für die Funktion an:

   INVESTMENT_INIT = 100000  # starting amount
   INVESTMENT_ANN = 10000  # yearly new investment
   TERM = 30  # number of years
   MKT_AVG_RETURN = 0.11 # percentage
   MKT_STD_DEV = 0.18  # standard deviation
   

7. Rufen Sie reduce ab, um die Werte im RDD zusammenzufassen. Geben Sie den folgenden Code ein, um die Ergebnisse im RDD zu summieren:

   sum = results.reduce(add)
   

8. Schätzen Sie die durchschnittliche Rendite ab und rufen Sie sie auf:

   print (sum / 10000.)
   

   Achten Sie darauf, den Punkt (.) am Ende einzufügen. Er kennzeichnet die Gleitkommaarithmetik.

9. Ändern Sie nun eine Annahme und sehen Sie, wie sich die Ergebnisse ändern. Sie können zum Beispiel einen neuen Wert für die durchschnittliche Rendite des Marktes eingeben:

   MKT_AVG_RETURN = 0.07
   

10. Führen Sie die Simulation noch einmal durch:

    print (sc.parallelize([time.time() + i for i in range(10000)]) \
            .map(grow).reduce(add)/10000.)
    

11. Wenn Sie fertig sind, drücken Sie CTRL+D, um den Python-Interpreter zu schließen.

Monte-Carlo-Simulation in Scala programmieren

Monte Carlo ist für seine Casinos bekannt. In diesem Abschnitt verwenden Sie Scala, um eine Simulation zu erstellen, die den mathematischen Vorteil zeigt, den ein Spielcasino in einem Glücksspiel genießt. Der "Hausvorteil" in einem echten Casino variiert von Spiel zu Spiel erheblich. Bei Keno kann er zum Beispiel über 20 % sein. Diese Anleitung erstellt ein einfaches Spiel, bei dem das Haus nur einen Vorteil von einem Prozent hat. So funktioniert das Spiel:

• Der Spieler platziert eine Wette, bestehend aus einer Reihe von Chips aus einem Banknotenbündel.
• Der Spieler rollt einen 100-seitigen Würfel.
• Wenn das Ergebnis dessen eine Zahl von 1 bis 49 ist, gewinnt der Spieler.
• Bei Ergebnissen von 50 bis 100 verliert der Spieler die Wette.

Sie sehen, dass dieses Spiel einen Nachteil für den Spieler in Höhe von einem Prozent schafft: Bei 51 der 100 möglichen Ergebnisse bei jedem Würfeln verliert der Spieler.

Gehen Sie so vor, um das Spiel zu erstellen und auszuführen:

1. Starten Sie den Scala-Interpreter über den primären Dataproc-Knoten.

   spark-shell
   

2. Kopieren und fügen Sie den folgenden Code ein, um das Spiel zu erstellen. Scala stellt bei der Einrückung nicht die gleichen Anforderungen wie Python, sodass Sie diesen Code einfach kopieren und in die Eingabeaufforderung scala> einfügen können.

   val STARTING_FUND = 10
   val STAKE = 1   // the amount of the bet
   val NUMBER_OF_GAMES = 25
   
   def rollDie: Int = {
       val r = scala.util.Random
       r.nextInt(99) + 1
   }
   
   def playGame(stake: Int): (Int) = {
       val faceValue = rollDie
       if (faceValue < 50)
           (2*stake)
       else
           (0)
   }
   
   // Function to play the game multiple times
   // Returns the final fund amount
   def playSession(
      startingFund: Int = STARTING_FUND,
      stake: Int = STAKE,
      numberOfGames: Int = NUMBER_OF_GAMES):
      (Int) = {
   
       // Initialize values
       var (currentFund, currentStake, currentGame) = (startingFund, 0, 1)
   
       // Keep playing until number of games is reached or funds run out
       while (currentGame <= numberOfGames && currentFund > 0) {
   
           // Set the current bet and deduct it from the fund
           currentStake = math.min(stake, currentFund)
           currentFund -= currentStake
   
           // Play the game
           val (winnings) = playGame(currentStake)
   
           // Add any winnings
           currentFund += winnings
   
           // Increment the loop counter
           currentGame += 1
       }
       (currentFund)
   }
   

3. Drücken Sie return, bis die Eingabeaufforderung scala> angezeigt wird.

4. Geben Sie den folgenden Code ein, um das Spiel 25-mal zu spielen. Dies ist der Standardwert für NUMBER_OF_GAMES.

   playSession()
   

   Ihre Banknotenrolle lag am Anfang bei einem Wert von 10 Einheiten. Ist er jetzt höher oder niedriger?

5. Simulieren Sie jetzt, dass 10.000 Spieler 100 Chips pro Spiel setzen. Spielen Sie 10.000 Spiele in einer Sitzung. Diese Monte-Carlo-Simulation berechnet die Wahrscheinlichkeit, mit der Sie Ihr Geld vor dem Ende der Sitzung verlieren. Geben Sie folgenden Code ein:

   (sc.parallelize(1 to 10000, 500)
     .map(i => playSession(100000, 100, 250000))
     .map(i => if (i == 0) 1 else 0)
     .reduce(_+_)/10000.0)
   

   Achten Sie darauf, dass die Syntax .reduce(_+_) eine Kurzschreibweise in Scala für die Zusammenfassung mithilfe einer Summenfunktion ist. Funktional gesehen entspricht sie der Syntax .reduce(add) aus dem Python-Beispiel.

   Dieser Code führt die folgenden Schritte aus:

   • Er erstellt einen RDD mit den Ergebnissen der Sitzung.
   • Er ersetzt die Ergebnisse der bankrotten Spieler durch die Zahl 1 und Ergebnisse über null durch die Zahl 0.
   • Er rechnet die Anzahl der bankrotten Spieler zusammen.
   • Er dividiert den Wert durch die Anzahl der Spieler.

   Ein typisches Ergebnis könnte so aussehen:

   0.998
   

   Dies ist fast eine Garantie dafür, dass Sie Ihr gesamtes Geld verlieren werden, obwohl das Casino nur einen Vorteil von einem Prozent hatte.