Connettore HDFS per lo storage degli oggetti

Per utilizzare il connettore HDFS, è necessario disporre dei seguenti elementi:

• Un account Oracle Cloud Infrastructure.
• Un utente creato in tale account, in un gruppo con un criterio che concede le autorizzazioni desiderate per qualsiasi bucket che si desidera utilizzare. Questo può essere un utente personale o un'altra persona/sistema che deve chiamare l'API. Per un esempio su come impostare un nuovo utente, gruppo, compartimento e criterio, vedere Aggiunta di utenti. Per un criterio di storage degli elementi di base, vedere Consenti agli amministratori di storage degli elementi di gestire bucket e oggetti.
• Java 8
• Valore TTL pari a 60. Per ulteriori informazioni, vedere Configurazione del TTL JVM per le ricerche dei nomi DNS.

Copiare i file jar in bundle da lib e da terze parti/lib in ogni nodo del cluster Hadoop in modo che siano inclusi nel CLASSPATH di Hadoop.

<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>

È possibile impostare le seguenti proprietà del connettore HDFS nel file core-site.xml. La pagina BmcProperties elenca le proprietà aggiuntive che è possibile configurare per una connessione allo storage degli oggetti.

Impostazione dell'endpoint area

Per impostare l'endpoint dell'area per il connettore HDFS è possibile utilizzare diversi metodi:

• Specifica della proprietà hostname in core-site.xml
• Specifica della proprietà codice area o identificativo area in core-site.xml
• Consentire al client ObjectStorage di selezionare l'endpoint tramite il servizio di metadati dell'istanza

Configurazione delle proprietà con core-site.xml

Questo esempio mostra come configurare le proprietà in un file core-site.xml (gli OCID vengono abbreviati per brevità):

<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 fornisce principali istanza in modo da non dover più configurare le credenziali utente o fornire file PEM sui servizi in esecuzione sulle istanze. Ognuna di queste istanze ha la propria identità e si esegue l'autenticazione utilizzando i certificati aggiunti all'istanza dai principal dell'istanza.

Per utilizzare l'autenticazione dei principal dell'istanza con il connettore HDFS, fornire semplicemente la proprietà fs.oci.client.custom.authenticator e impostare il valore su com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator.

Poiché l'uso dei principal dell'istanza fornisce al connettore un autenticatore personalizzato, non è più necessario configurare le proprietà riportate di seguito.

• 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

Il seguente codice di esempio illustra l'uso dei principal dell'istanza per l'autenticazione con il connettore HDFS:

<?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>

Per ulteriori informazioni sui principal delle istanze, vedere Annuncio dei principal delle istanze per Identity and Access Management.

Analogamente ai principal delle istanze, Oracle fornisce principali risorse per autenticare le risorse che non sono istanze, ad esempio un notebook jupyter. Ogni risorsa ha la propria identità ed esegue l'autenticazione utilizzando i certificati aggiunti.

Per utilizzare l'autenticazione dei principal risorsa con il connettore HDFS, fornire semplicemente la proprietà fs.oci.client.custom.authenticator e impostare il valore su com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator.

Poiché l'utilizzo dei principal risorsa fornisce al connettore un autenticatore personalizzato, non è necessario configurare a lungo le seguenti proprietà:

• 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

Il seguente codice di esempio illustra l'uso dei principal risorsa per l'autenticazione con il connettore HDFS:

<?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>

Per ulteriori informazioni sui principal delle istanze, vedere Utilizzo dei principal delle risorse nel servizio Data Science.

Oracle supporta l'autenticazione Kerberos per connettersi allo storage degli oggetti utilizzando il connettore HDFS.

Il file core-site.xml di esempio seguente illustra l'uso di Kerberos con l'autenticazione del token SPNEGO con il connettore HDFS.

<?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>

Per ulteriori informazioni su Kerberos, vedere Esercitazione sul protocollo Kerberos.

Per ulteriori informazioni su SPNEGO, vedere RFC 4178.

A partire dalla versione 3.3.0.7.0.0, HDFS supporta l'uso del client Apache per impostazione predefinita per effettuare chiamate di servizio OCI. Questo perché HDFS Connector si basa sull'SDK per Java per inviare richieste al server. L'SDK per Java supporta l'uso di Jersey ApacheConnectorProvider per impostazione predefinita anziché Jersey HttpUrlConnectorProvider per consentire a Apache HttpClient di effettuare chiamate ai servizi OCI.

Per tornare al vecchio client predefinito di Jersey, impostare la proprietà fs.oci.client.jersey.default.connector.enabled nel file core-site.xml su true. Per impostazione predefinita, questo valore viene impostato su false.

È possibile impostare il numero massimo di connessioni nel connection pool del connettore HDFS.

A tale scopo, modificare la proprietà fs.oci.client.apache.max.connection.pool.size nel file core-site.xml in un numero intero positivo che specifica il numero di connessioni al pool.

È possibile impostare le seguenti proprietà opzionali nel file core-site.xml per configurare un proxy HTTP:

Gli oggetti di grandi dimensioni vengono caricati nello storage degli oggetti utilizzando caricamenti multiparte. Il file viene diviso in parti più piccole che vengono caricate in parallelo, riducendo così i tempi di caricamento. Ciò consente inoltre al connettore HDFS di riprovare a eseguire il caricamento delle parti non riuscite invece di non riuscire nell'intero caricamento. Tuttavia, i caricamenti potrebbero non riuscire in modo temporaneo e il connettore tenterà di interrompere i file caricati parzialmente. Poiché questi file si accumulano (e ti verrà addebitato il costo per lo storage), elencare i caricamenti periodicamente e quindi dopo un certo numero di giorni interromperli manualmente utilizzando l'SDK per Java.

Le informazioni sull'uso dell'API di storage degli oggetti per la gestione dei caricamenti multiparte sono disponibili in Utilizzo dei caricamenti multiparte.

Le sezioni seguenti contengono procedure ottimali per ottimizzare l'uso e le prestazioni.

Si fa riferimento ai file system e ai file HDFS tramite gli URI. Lo schema specifica il tipo di file system e la parte rimanente dell'URI è in gran parte libera per l'implementazione del file system da interpretare come vuole.

Poiché lo storage degli oggetti è un'area di memorizzazione degli oggetti, la relativa capacità di assegnare un nome agli oggetti come se fossero file in un file system viene utilizzata per imitare un file system effettivo.

oci://MyBucket@MyNamespace/
		

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

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

Il login al connettore viene eseguito tramite SLF4J. SLF4J è un'astrazione di log che consente l'uso di una libreria di log fornita dall'utente (ad esempio, log4j). Per ulteriori informazioni, vedere il manuale SLF4J.

L'esempio seguente mostra come abilitare il log di base all'output standard.

1. Scaricare il file jar di associazione semplice SLF4J: SLF4J Simple Binding
2. Aggiungi il file jar al classpath
3. Aggiungere l'argomento VM seguente per abilitare il log del livello di debug (per impostazione predefinita, viene utilizzato il livello di informazioni): -Dorg.slf4j.simpleLogger.defaultLogLevel=debug

È possibile configurare opzioni di log più avanzate utilizzando l'associazione log4j.

Il connettore HDFS per lo storage degli oggetti include un framework di monitoraggio che fornisce metriche sulle operazioni eseguite utilizzando il connettore. Il framework di monitoraggio fornisce un'interfaccia che può essere implementata per utilizzare/ascoltare le metriche generate dal connettore. È possibile fornire un'implementazione personalizzata di questa interfaccia oppure utilizzare l'implementazione della telemetria pubblica OCI inclusa in questo framework.

Guida introduttiva

Per iniziare a utilizzare il framework di monitoraggio del connettore HDFS, sarà necessario impostare le seguenti proprietà. Una volta impostate queste proprietà per OCIMonitoring, è possibile utilizzare la vista Explorer metriche nella console OCI per osservare le metriche emesse dal connettore HDFS.

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>

Proprietà com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin

Se la proprietà com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin è abilitata, sono applicabili le proprietà riportate di seguito.

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>

Creazione del proprio consumer per le metriche

L'estensione delle classi deve avere un costruttore con la stessa firma della classe OCIMonitorConsumerPlugin.

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();
}

La classe OCIMetric può essere implementata in tre modi:

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);
    }
}

Questa sezione contiene informazioni sulla risoluzione dei problemi per il connettore HDFS.

Hai trovato una soluzione per un bug o una nuova funzionalità a cui vorresti contribuire? L'SDK è open source e accetta le richieste Pull su GitHub.

Se si desidera ricevere una notifica quando viene rilasciata una nuova versione del connettore HDFS, effettuare la sottoscrizione al feed Atom.

Modi per entrare in contatto:

• GitHub Problemi: per segnalare bug ed effettuare richieste di funzioni
• Overflow dello stack: utilizzare i tag oracle-cloud-infrastructure e oci-hdfs-connector nel post
• Sezione Strumenti per sviluppatori dei forum di Oracle Cloud
• My Oracle Support