[PYTHON] Ein Hinweis zum Verbinden von Spark mit OpenStack Swift-basiertem IBM Object Storage

Obwohl es verschiedene Datenquellen gibt, scheint die Hürde gering zu sein, um die Datei in Eile in den Speicher zu stellen. Wenn es sich um eine Big-Data-Prämisse wie Spark handelt, werden häufig Protokolldateien usw. angenommen, aber da allgemeine Geschäftsdaten nicht vermieden werden können, zuerst CSV, also als ich ein wenig CSV ausprobiert habe Hinweis. (Bei IBM Data Scientist Experience Python 2 mit Spark 2.0)

Vorbereitung

Laden Sie die CSV-Datei in den Swift-basierten IBM Object Storage hoch. Verwenden Sie "Legen Sie Ihre Datei hier ab oder durchsuchen Sie Ihre Dateien, um eine neue Datei hinzuzufügen" auf der rechten Seite des Bildschirms. Ich habe Baseball.csv, cars.csv, Whisky.csv hochgeladen, wie im Screenshot gezeigt.

Ich habe versucht, Daten aus dem Objektspeicher zu lesen

Zuerst habe ich es mit dem Code versucht, der aus dem DSX-Code eingefügt wurde

Es gibt einen Prozess zum Festlegen der Konfiguration und einen Prozess zum Laden von CSV-Daten in DataFrame. (1) Stellen Sie die Hadoop-Konfiguration über Spark ein Die unter Zugriff auf OpenStack Swift von Spark beschriebenen Konfigurationsparameter werden gemäß der in IBM Object Storage hochgeladenen Datei festgelegt. Es scheint, dass

test.py

# @hidden_cell
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

# This function accesses a file in your Object Storage. The definition contains your credentials.
# You might want to remove those credentials before you share (name):
def set_hadoop_config_with_credentials_xxxxxxxxxxxxxxxxxxcxxxxxcc(name):
    """This function sets the Hadoop configuration so it is possible to
    access data from Bluemix Object Storage V3 using Spark"""

    prefix = 'fs.swift.service.' + name
    hconf = sc._jsc.hadoopConfiguration()
    hconf.set(prefix + '.auth.url', 'https://identity.open.softlayer.com'+'/v3/auth/tokens')
    hconf.set(prefix + '.auth.endpoint.prefix', 'endpoints')
    hconf.set(prefix + '.tenant', '4dbbbca3a8ec42eea9349120fb91dcf9')
    hconf.set(prefix + '.username', 'xxxxxxxxxcxxxxxxcccxxxxxccc')
    hconf.set(prefix + '.password', 'Xxxxxxxxxxxxxxxxx')
    hconf.setInt(prefix + '.http.port', 8080)
    hconf.set(prefix + '.region', 'dallas')
    hconf.setBoolean(prefix + '.public', True)

# you can choose any name
name = 'keystone'
set_hadoop_config_with_credentials_Xxxxxxxxxxxxxxxxxxcxxxxxc(name)

(2) Laden Sie CSV-Daten in den Spark-Datenrahmen

test.py

df_data_1=sqlContext.read.format('com.databricks.spark.csv')\
    .options(header='true',inferschema='true')\
    .load("swift://PredictiveAnalyticsProject2." + name + "/whiskey.csv")

Dies scheint ein Paket namens spark-csv zu verwenden, das vor Spark 2.0 benötigt wurde. Ab spark2.0 scheint Spark DataFrame in der Lage zu sein, csv direkt zu verarbeiten, und read.csv () kann ebenfalls verwendet werden, was praktisch ist.

Ich habe versucht, Daten in Object Storage zu schreiben

Der Inhalt des Spark DataFrame, der whisky.csv geladen hat, wurde unter dem Namen whisky_new.csv in den Objektspeicher exportiert. (1) Ich habe versucht, mit Schreiben von Spark Data Frame auszugeben Schreiben Sie einfach write.csv () und es ist OK.

Was den Modus betrifft, scheint der Speichermodus wie in hier gezeigt verwendet werden zu können. Die Ausgabedatei wurde in mehrere Dateien unterteilt und anstelle einer CSV ausgegeben, wie im folgenden Screenshot gezeigt. Ich denke, das liegt daran, dass es von mehreren Knoten verarbeitet wird. (Es ist seltsam? Wenn Sie dies mit read.csv erneut von Spark lesen, wird es als eine Textdatei, csv, behandelt, sodass es kein Problem zu geben scheint.) Es scheint jedoch, dass es andere Personen als mich gibt, die in einer Datei ausgeben möchten, und es wurde auch gefragt, ob die Ausgabe in einer Datei möglich ist (http://stackoverflow.com/questions/31674530/write-single-). CSV-Datei mit Spark-CSV)

(2) Ich habe die REST-API ohne Verwendung von Spark ausprobiert Ich denke nicht, dass es für die Verarbeitung einer großen Datenmenge geeignet ist, aber Sie können es mit dem folgenden Code als eine CSV im Objektspeicher ablegen. (Der get-Teil des Codes, den DSX zum Erstellen von Pandas DataFrame einfügt, wird nur in put geändert. Geben Sie target / url2 zum Hochladen aus der Antwort ein, während Sie die Authentifizierungsinformationen abrufen, indem Sie den ersten API-Aufruf / POST ausführen Scheint zusammenzubauen)

put_sample.py

def put_object_storage_file_with_credentials_xxxxxxxxxx(container, filename, indata):
    url1 = ''.join(['https://identity.open.softlayer.com', '/v3/auth/tokens'])
    data = {'auth': {'identity': {'methods': ['password'],
            'password': {'user': {'name': 'member_1825cd3bc875420fc629ccfd22c22e20433a7ac9','domain': {'id': '07e33cca1abe47d293b86de49f1aa8bc'},
            'password': 'xxxxxxxxxx'}}}}}
    headers1 = {'Content-Type': 'application/json'}
    resp1 = requests.post(url=url1, data=json.dumps(data), headers=headers1)
    resp1_body = resp1.json()
    for e1 in resp1_body['token']['catalog']:
        if(e1['type']=='object-store'):
            for e2 in e1['endpoints']:
                        if(e2['interface']=='public'and e2['region']=='dallas'):
                            url2 = ''.join([e2['url'],'/', container, '/', filename])
    s_subject_token = resp1.headers['x-subject-token']
    headers2 = {'X-Auth-Token': s_subject_token, 'accept': 'application/json'}
    resp2 = requests.put(url=url2, headers=headers2 , data=indata)
    print resp2
    return (resp2.content)


put_object_storage_file_with_credentials_xxxxxxxxxx( 'PredictiveAnalyticsProject2', 'whiskey_new_s.csv' , df_data_2.to_csv( index = False ) )

Nachtrag

Es scheint eine Möglichkeit zu geben, Pythons Swiftclient als Image für die obige REST-API zu verwenden, aber es schien in der Data Scientist Experience-Umgebung von IBM nicht zu funktionieren. Using IBM Object Storage in Bluemix, with Python