HDFS-Connector für Object Storage

Zur Verwendung des HDFS-Connectors ist Folgendes erforderlich:

• Ein Oracle Cloud Infrastructure-Account.
• Ein in diesem Account erstellter Benutzer in einer Gruppe mit einer Policy, die die gewünschten Berechtigungen für jeden zu verwendenden Bucket erteilt. Dies kann ein Benutzer für Sie selbst oder eine andere Person/ein anderes System sein, die/das die API aufrufen muss. Ein Beispiel für die Einrichtung eines neuen Benutzers, einer neuen Gruppe, eines neuen Compartments und einer neuen Policy finden Sie unter Benutzer hinzufügen. Informationen zu einer allgemeinen Object Storage-Policy finden Sie unter Verwalten von Buckets und Objekten durch Object Storage-Administratoren zulässig.
• Java 8
• Ein TTL-Wert von 60. Weitere Informationen finden Sie unter JVM-TTL für DNS-Namen-Lookups konfigurieren.

Kopieren Sie die gebündelten JAR-Dateien aus lib und third-party/lib in jeden Knoten des Hadoop-Clusters, damit sie im Classpath von Hadoop enthalten sind.

<dependency>
  <groupId>com.oracle.oci.sdk</groupId>
  <artifactId>oci-hdfs-connector</artifactId>
  <!-- Replace the version below with your required version -->
  <version>2.9.2.0</version>
</dependency>

Sie können die folgenden HDFS-Connector-Eigenschaften in der Datei core-site.xml festlegen. Auf der Seite BmcProperties werden zusätzliche Eigenschaften aufgeführt, die Sie für eine Verbindung zu Object Storage konfigurieren können.

Regionsendpunkt festlegen

Es gibt verschiedene Methoden, mit denen Sie den Regionsendpunkt für den HDFS-Connector festlegen können:

• Hostnamenseigenschaft in core-site.xml angeben
• Regionscode oder Regions-ID-Eigenschaft in core-site.xml angeben
• Zulassen, dass der ObjectStorage-Client den Endpunkt über den Instance Metadata Service abruft

Eigenschaften mit core-site.xml konfigurieren

Dieses Beispiel zeigt, wie Eigenschaften in einer core-site.xml-Datei konfiguriert werden können (OCIDs sind aus Gründen der Einfachheit gekürzt):

<configuration>
...
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-ashburn-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.hostname.myBucket.myNamespace</name>
    <value>https://objectstorage.us-phoenix-1.oraclecloud.com</value><!-- Use Phoenix for myBucket@myNamespace -->
  </property>
  <property>
    <name>fs.oci.client.auth.tenantId</name>
    <value>ocid1.tenancy.oc1..exampleuniqueID</value> 
  </property>
  <property>
    <name>fs.oci.client.auth.userId</name>
    <value>ocid1.user.oc1..exampleuniqueID</value>
  </property>
  <property>
    <name>fs.oci.client.auth.fingerprint</name>
    <value>20:3b:97:13:55:1c:5b:0d:d3:37:d8:50:4e:c5:3a:34</value>
  </property>
  <property>
    <name>fs.oci.client.auth.pemfilepath</name>
    <value>~/.oci/oci_api_key.pem</value>
  </property>
...
</configuration>

Oracle stellt Instanz-Principals bereit, damit Sie keine Benutzerzugangsdaten mehr konfigurieren oder PEM-Dateien auf Services angeben müssen, die auf Instanzen ausgeführt werden. Jede dieser Instanzen hat eine eigene Identität und authentifiziert sich durch Verwendung von Zertifikaten, die der Instanz von Instanz-Principals hinzugefügt wurden.

Um die Authentifizierung durch Instanz-Principals mit dem HDFS-Connector zu verwenden, geben Sie die Eigenschaft fs.oci.client.custom.authenticator an, und setzen Sie den Wert auf com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator.

Da die Verwendung von Instanz-Principals dem Connector einen benutzerdefinierten Authentifizierer bereitstellt, müssen die folgenden Eigenschaften nicht mehr konfiguriert werden:

• fs.oci.client.auth.tenantId
• fs.oci.client.auth.userId
• fs.oci.client.auth.fingerprint
• fs.oci.client.auth.pemfilepath
• fs.oci.client.auth.passphrase

Der folgende Beispielcode zeigt, wie Instanz-Principals zur Authentifizierung mit dem HDFS-Connector verwendet werden:

<?xml version="1.0"?>
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-phoenix-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.custom.authenticator</name>
    <value>com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator</value>
  </property>
</configuration>

Weitere Informationen zu Instanz-Principals finden Sie unter Instanz-Principals für Identity and Access Management ankündigen.

Ähnlich wie Instanz-Principals stellt Oracle Resource Principals zur Authentifizierung der Ressourcen zur Verfügung, die keine Instanzen sind (wie ein Jupyter-Notizbuch). Jede Ressource hat eine eigene Identität und authentifiziert sich mit den ihr hinzugefügten Zertifikaten.

Um die Authentifizierung durch Resource Principals mit dem HDFS-Connector zu verwenden, geben Sie die Eigenschaft fs.oci.client.custom.authenticator an, und setzen Sie den Wert auf com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator.

Da die Verwendung von Resource Principals dem Connector einen benutzerdefinierten Authentifizierer bereitstellt, ist es nicht lange erforderlich, die folgenden Eigenschaften zu konfigurieren:

• fs.oci.client.auth.tenantId
• fs.oci.client.auth.userId
• fs.oci.client.auth.fingerprint
• fs.oci.client.auth.pemfilepath
• fs.oci.client.auth.passphrase

Der folgende Beispielcode zeigt, wie Resource Principals zur Authentifizierung mit dem HDFS-Connector verwendet werden:

<?xml version="1.0"?>
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-phoenix-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.custom.authenticator</name>
    <value>com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator</value>
  </property>
</configuration>

Weitere Informationen zu Instanz-Principals finden Sie unter Resource Principals im Data Science-Service verwenden.

Oracle unterstützt die Kerberos-Authentifizierung für die Verbindung mit Object Storage über den HDFS-Connector.

Die folgende Beispieldatei core-site.xml veranschaulicht die Verwendung von Kerberos mit SPNEGO-Tokenauthentifizierung mit dem HDFS-Connector:

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-phoenix-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.auth.tenantId</name>
    <value></value>
  </property>
  <property>
    <name>fs.oci.client.auth.userId</name>
    <value></value>
  </property>
   <property>
    <name>fs.oci.client.custom.authenticator</name>
    <value>com.oracle.bmc.hdfs.auth.spnego.UPSTAuthenticationCustomAuthenticator</value>
  </property>
  <property>
    <name>fs.oci.client.upst.tokenExchangeServicePrincipal</name>
    <value><!-- Service Principal for generating SPNEGO token --></value>
  </property>
  <property>
    <name>fs.oci.client.upst.userPrincipal</name>
    <value><!-- User Principal for generating SPNEGO token --></value>
  </property>
  <property>
    <name>fs.oci.client.upst.issuer</name>
    <value><!-- Issuer for generating SPNEGO token --></value>
  </property>
  <property>
    <name>fs.oci.client.upst.domainUrl</name>
    <value><!-- Domain application client ID for IAM token exchange --></value>
  </property>
  <property>
    <name>fs.oci.client.upst.clientId</name>
    <value><!-- Domain application client ID for IAM token exchange --></value>
  </property>
  <property>
    <name>fs.oci.client.upst.clientSecret</name>
    <value><!-- Domain application client secret for IAM token exchange --></value>
  </property>
  <property>
    <name>fs.oci.client.keytab.path</name>
    <value><!-- File path to the keytab used for token exchange --></value>
  </property>
 </configuration>

Weitere Informationen zu Kerberos finden Sie im Kerberos-Protokoll-Tutorial.

Weitere Informationen zu SPNEGO finden Sie unter RFC 4178.

Ab Version 3.3.0.7.0.0 unterstützt HDFS standardmäßig die Verwendung des Apache-Clients zum Ausführen von OCI-Serviceaufrufen. Der Grund ist, dass sich der HDFS-Connector auf das SDK für Java verlässt, um Anforderungen an den Server zu senden. Das SDK für Java unterstützt standardmäßig die Verwendung von Jersey ApacheConnectorProvider anstelle von Jersey HttpUrlConnectorProvider, damit Apache HttpClient OCI-Serviceaufrufe ausführen kann.

Um zum alten Jersey-Standardclient zurückzukehren, setzen Sie die Eigenschaft fs.oci.client.jersey.default.connector.enabled in der Datei core-site.xml auf true. Standardmäßig ist dieser Wert auf false gesetzt.

Sie können die maximale Anzahl von Verbindungen im HDFS-Connector-Verbindungspool festlegen.

Ändern Sie dazu die Eigenschaft fs.oci.client.apache.max.connection.pool.size in der Datei core-site.xml in eine positive Ganzzahl, mit der Sie die Anzahl der Verbindungen im Pool angeben.

Sie können die folgenden optionalen Eigenschaften in der Datei core-site.xml festlegen, um einen HTTP-Proxy zu konfigurieren:

Large Objects werden mithilfe von Multipart-Uploads in Object Storage hochgeladen. Die Datei wird in kleinere Teile aufgeteilt, die parallel hochgeladen werden, wodurch die Uploadzeiten reduziert werden. Dadurch kann der HDFS-Connector zudem den Upload nicht erfolgreicher Teile wiederholen, statt dass der gesamte Upload nicht erfolgreich verläuft. Uploads können jedoch vorübergehend nicht erfolgreich sein, und der Connector versucht, teilweise hochgeladene Dateien abzubrechen. Da sich diese Dateien ansammeln (und Sie für die Speicherung bezahlen müssen), listen Sie die Uploads regelmäßig und danach nach einer bestimmten Anzahl von Tagen manuell mit dem SDK für Java auf.

Informationen zur Verwendung der Object Storage-API zur Verwaltung von mehrteiligen Uploads finden Sie unter Multipart-Uploads verwenden.

Die folgenden Abschnitte enthalten Best Practices zur Optimierung von Nutzung und Performance.

HDFS-Dateisysteme und -Dateien werden über URIs referenziert. Das Schema gibt den Dateisystemtyp an, und der verbleibende Teil der URI ist größtenteils frei, sodass die Dateisystemimplementierung ihn je nach Bedarf interpretieren kann.

Da Object Storage ein Objektspeicher ist, wird dessen Fähigkeit, Objekte so zu benennen, als wären es Dateien in einem Dateisystem, verwendet, um ein tatsächliches Dateisystem nachgeahmt.

oci://MyBucket@MyNamespace/
		

oci://MyBucket@MyNamespace/a/b/
		

oci://MyBucket@MyNamespace/a/b/c/example.json
		

Das Logging im Connector erfolgt über SLF4J. SLF4J ist eine Loggingabstraktion, die die Verwendung einer vom Benutzer angegebenen Logging-Library (z.B. log4j) ermöglicht. Weitere Informationen finden Sie im SLF4J-Handbuch.

Das folgende Beispiel zeigt, wie das allgemeine Logging für die Standardausgabe aktiviert wird.

1. Laden Sie die SLF4J-JAR-Datei für einfaches Binding herunter: SLF4J für einfaches Binding
2. Fügen Sie die JAR-Datei dem Classpath hinzu.
3. Fügen Sie das folgende VM-Argument hinzu, um das Logging auf Debugebene zu aktivieren (standardmäßig wird die Informationsebene verwendet): -Dorg.slf4j.simpleLogger.defaultLogLevel=debug

Mit dem log4j-Binding können Sie erweiterte Loggingoptionen konfigurieren.

Der HDFS-Connector für Object Storage umfasst ein Monitoring-Framework, das Metriken für Vorgänge bereitstellt, die mit dem Connector ausgeführt werden. Das Monitoring-Framework bietet eine Schnittstelle, die implementiert werden kann, um vom Connector generierte Metriken zu konsumieren/zuhören. Sie können eine benutzerdefinierte Implementierung dieser Schnittstelle bereitstellen oder die öffentliche OCI-Telemetrieimplementierung verwenden, die in diesem Framework enthalten ist.

Erste Schritte

Um das HDFS Connector-Überwachungs-Framework zu verwenden, müssen Sie die folgenden Eigenschaften festlegen. Nachdem diese Eigenschaften für OCIMonitoring festgelegt wurden, können Sie die Metriken, die vom HDFS-Connector ausgegeben werden, in der Ansicht "Metrik-Explorer" in der OCI-Konsole beobachten.

fs.oci.mon.consumer.plugins

<property>
    <name>fs.oci.mon.consumer.plugins</name>
    <value>com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin,com.your.new.plugin.PluginImpl1</value>
</property>

fs.oci.mon.grouping.cluster.id

<property>
    <name>fs.oci.mon.grouping.cluster.id</name>
    <value>hdfs-sample-cluster-id</value>
</property>

com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin-Eigenschaften

Wenn die Eigenschaft com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin aktiviert ist, gelten die folgenden Eigenschaften:

fs.oci.mon.telemetry.ingestion.endpoint

<property>
    <name>fs.oci.mon.telemetry.ingestion.endpoint</name>
    <value>https://telemetry-ingestion.us-ashburn-1.oraclecloud.com</value>
</property>

fs.oci.mon.compartment.ocid

<property>
    <name>fs.oci.mon.compartment.ocid</name>
    <value>ocid1.compartment.oc1..sample.compartment.id</value>
</property>

fs.oci.mon.bucket.level.enabled

<property>
    <name>fs.oci.mon.bucket.level.enabled</name>
    <value>true</value>
</property>

fs.oci.mon.ns.name

<property>
    <name>fs.oci.mon.ns.name</name>
    <value>name.of.namespace.on.oci.telemetry</value>
</property>

fs.oci.mon.rg.name

<property>
    <name>fs.oci.mon.rg.name</name>
    <value>name.of.resource.group.on.oci.telemetry</value>
</property>

Eigenen Consumer für Metriken erstellen

Erweiterungsklassen sollten einen Konstruktor mit derselben Signatur wie die Klasse OCIMonitorConsumerPlugin haben.

public OCIMonitorConsumerPlugin(BmcPropertyAccessor propertyAccessor, String bucketName, String monitoringGroupingID, String namespaceName);

/**
 * This class that has to be extended by any plugin, that wants to consume the metrics emitted by OCI HDFS connector.
 */
public abstract class OCIMonitorConsumerPlugin {
    /**
     * This method will be called on each plugin, by the OCI monitoring framework, whenever it wants to emit out a metric.
     * This method should finish as quickly as possible, so the consumer of this should ideally handover the
     * ocimetric and stage it elsewhere for processing, instead of trying to deal with it in the accept call itself.
     * @param ociMetric The metric that is being emitted by the OCI HDFS connector
     */
    public void accept(OCIMetric ociMetric);
 
    /**
     * This shutdown method will be called on the implementing plugins, whenever the JVM is shutting down.
     * It could be used to cleanup, finish pending tasks before exit.
     */
    public void shutdown();
}

Die Klasse OCIMetric kann auf drei Arten implementiert werden:

public class OCIMetric {
    /**
     * The time in milliseconds (epoch) when the metric was recorded.
     */
    private final long recordedTime;
    /**
     * The overall time taken by the operation to complete/error-out in milliseconds.
     */
    private final double overallTime;
    /**
     * The operation key. This will be one of {"LIST", "HEAD", "WRITE", "READ", "DELETE", "RENAME"}
     */
    private final String key;
    /**
     * The boolean error indicates whether the operation errored out.
     */
    private final boolean error;
    /**
     * The target OCI bucket where the operation was attempted to.
     */
    private final String bucketName;
}

public class OCIMetricWithThroughput extends OCIMetric {
    /**
     * The throughput that was recorded for the operation in bytes/second
     */
    private final double throughput;
    /**
     * The total count of bytes that were transferred in or out.
     */
    private final double bytesTransferred;
}

public class OCIMetricWithFBLatency extends OCIMetricWithThroughput {
    /**
     * The time to first byte when a read operation was performed in milliseconds.
     */
    private final double ttfb;
}

hadoop_sample_hdfs:

package com.oracle.oci.hadoop.example;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.oracle.oci.hdfs.BmcFilesystem;

import lombok.RequiredArgsConstructor;

@RequiredArgsConstructor
public class SampleOracleBmcHadoopJob
{
    private static final String SAMPLE_JOB_PATH = "/samplehadoopjob";
    private static final String INPUT_FILE = SAMPLE_JOB_PATH + "/input.dat";
    private static final String OUTPUT_DIR = SAMPLE_JOB_PATH + "/output";

    // non-static since this is the runner class it needs to initialize after we set the properties
    private final Logger log = LoggerFactory.getLogger(SampleOracleBmcHadoopJob.class);

    /**
     * Runner for sample hadoop job. This expects 3 args: path to configuration file, Object Store namespace, Object
     * Store bucket. To run this, you must:
     *{@code 


         * 
    Create a standard hadoop configuration file

         * 
    Create the bucket ahead of time.

         *} 


     * This runner will create a test input file in a file '/samplehadoopjob/input.dat', and job results will be written
     * to '/samplehadoopjob/output'.
     * 
     * @param args
     *            1) path to configuration file, 2) namespace, 3) bucket
     * @throws Exception
     */
    public static void main(final String[] args) throws Exception
    {
        if (args.length != 3)
        {
            throw new IllegalArgumentException(
                    "Must have 3 args: 1) path to config file, 2) object storage namespace, 3) object storage bucket");
        }

        // redirect all logs to sysout
        System.setProperty("org.slf4j.simpleLogger.logFile", "System.out");
        System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "debug");

        final SampleOracleBmcHadoopJob job = new SampleOracleBmcHadoopJob(args[0], args[1], args[2]);
        System.exit(job.execute());
    }

    private final String configurationFilePath;
    private final String namespace;
    private final String bucket;

    public int execute() throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException
    {
        log.info("Creating hadoop configuration");
        final Configuration configuration = this.createConfiguration(this.configurationFilePath);

        final String authority = this.bucket + "@" + this.namespace;
        final String uri = "oci://" + authority;
        log.info("Using uri: {}", uri);

        log.info("Creating job inputs");
        this.setup(uri, configuration);

        log.info("Creating job");
        final Job job = this.createJob(configuration);

        final String in = uri + INPUT_FILE;
        final String out = uri + OUTPUT_DIR;
        log.info("Using input: {}", in);
        log.info("Using output: {}", out);

        FileInputFormat.addInputPath(job, new Path(in));
        FileOutputFormat.setOutputPath(job, new Path(out));

        log.info("Executing job...");
        final int response = job.waitForCompletion(true) ? 0 : 1;

        log.info("Attempting to read job results");
        this.tryReadResult(uri, configuration);
        return response;
    }

    private Configuration createConfiguration(final String configFilePath)
    {
        final Configuration configuration = new Configuration();
        configuration.addResource(new Path(configFilePath));
        return configuration;
    }

    private void setup(final String uri, final Configuration configuration) throws IOException, URISyntaxException
    {
        try (final BmcFilesystem fs = new BmcFilesystem())
        {
            fs.initialize(new URI(uri), configuration);
            fs.delete(new Path(SAMPLE_JOB_PATH), true);
            final FSDataOutputStream output = fs.create(new Path(INPUT_FILE));
            output.writeChars("example\npath\ngak\ntest\nexample\ngak\n\ngak");
            output.close();
        }
    }

    private Job createJob(final Configuration configuration) throws IOException
    {
        final Job job = Job.getInstance(configuration, "word count");
        job.setJarByClass(SampleOracleBmcHadoopJob.class);
        job.setMapperClass(SimpleMapper.class);
        job.setCombinerClass(SimpleReducer.class);
        job.setReducerClass(SimpleReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        return job;
    }

    private void tryReadResult(final String uri, final Configuration configuration)
            throws IOException, URISyntaxException
    {
        try (final BmcFilesystem fs = new BmcFilesystem())
        {
            fs.initialize(new URI(uri), configuration);
            // this should be the output file name, but that could change
            final FSDataInputStream input = fs.open(new Path(OUTPUT_DIR + "/part-r-00000"));

            final ByteArrayOutputStream baos = new ByteArrayOutputStream();
            IOUtils.copy(input, baos);
            log.info("\n=====\n" + baos.toString() + "=====");
            input.close();
        }
    }
}


package com.oracle.oci.hadoop.example;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SimpleMapper extends Mapper
{
    private final static IntWritable one = new IntWritable(1);
    private final Text word = new Text();

    @Override
    public void map(final Object key, final Text value, final Context context) throws IOException, InterruptedException
    {
        final StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens())
        {
            this.word.set(itr.nextToken());
            context.write(this.word, one);
        }
    }
}


package com.oracle.oci.hadoop.example;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SimpleReducer extends Reducer
{
    private final IntWritable result = new IntWritable();

    @Override
    public void reduce(final Text key, final Iterable values, final Context context)
            throws IOException, InterruptedException
    {
        int sum = 0;
        for (final IntWritable val : values)
        {
            sum += val.get();
        }
        this.result.set(sum);
        context.write(key, this.result);
    }
}

Dieser Abschnitt enthält Informationen zur Fehlerbehebung für den HDFS-Connector.

Sie haben einen Bugfix oder ein neues Feature, das Sie beitragen möchten? Das SDK ist Open Source und akzeptiert Pull-Anforderungen auf GitHub.

Wenn Sie benachrichtigt werden möchten, wenn eine neue Version des HDFS-Connectors veröffentlicht wird, abonnieren Sie den Atom-Feed.

Kontaktmöglichkeiten:

• GitHub-Probleme: Zum Einreichen von Bugs und Featureanfragen
• Stack Overflow: Verwenden Sie die Tags oracle-cloud-infrastructure und oci-hdfs-connector in Ihrem Post.
• Abschnitt "Entwicklertools" der Oracle Cloud-Foren
• My Oracle Support