Connecteur HDFS pour Object Storage

Pour utiliser le connecteur HDFS, vous devez disposer des éléments suivants :

• Un compte Oracle Cloud Infrastructure.
• Un utilisateur créé dans ce compte, dans un groupe avec une stratégie qui octroie les droits d'accès souhaités sur les buckets à utiliser. Il peut s'agir d'un utilisateur pour vous-même ou d'une autre personne/un autre système devant appeler l'API. Pour obtenir un exemple de configuration d'un nouvel utilisateur, d'un nouveau groupe, d'un nouveau compartiment et d'une nouvelle stratégie, reportez-vous à Ajout d'utilisateurs. Pour obtenir une stratégie Object Storage de base, reportez-vous à Autoriser l'administration Object Storage à gérer les buckets et les objets.
• Java 8
• Une valeur de durée de vie de 60. Pour plus d'informations, reportez-vous à Configuration de la durée de vie de la JVM pour les recherches de noms DNS.

Copiez les fichiers JAR regroupés à partir de lib et de third-party/lib vers chaque noeud du cluster Hadoop afin qu'ils soient inclus dans la variable d'environnement CLASSPATH de 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>

Vous pouvez définir les propriétés de connecteur HDFS suivantes dans le fichier core-site.xml. La page BmcProperties répertorie des propriétés supplémentaires que vous pouvez configurer pour une connexion à Object Storage.

Définition de l'adresse de région

Vous pouvez utiliser plusieurs méthodes afin de définir l'adresse de région pour le connecteur HDFS :

• Spécification de la propriété de nom d'hôte dans core-site.xml
• Spécification de la propriété de code de région ou d'identificateur de région dans core-site.xml
• Autorisation de récupération de l'adresse via le service de métadonnées d'instance par le client ObjectStorage

Configuration de propriétés avec core-site.xml

Cet exemple montre comment les propriétés peuvent être configurées dans un fichier core-site.xml (les OCID sont abrégés dans un souci de concision) :

<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 fournit des principaux d'instance afin que vous n'ayez plus à configurer des informations d'identification utilisateur ou à fournir des fichiers PEM sur les services exécutés sur des instances. Chacune de ces instances possède sa propre identité et s'authentifie à l'aide de certificats ajoutés à l'instance par les principaux d'instance.

Pour utiliser l'authentification via un principal d'instance avec le connecteur HDFS, indiquez simplement la propriété fs.oci.client.custom.authenticator et définissez la valeur sur com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator.

Etant donné que l'utilisation de principaux d'instance fournit au connecteur un authentificateur personnalisé, il n'est plus nécessaire de configurer les propriétés suivantes :

• 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

L'exemple de code suivant illustre l'utilisation de principaux d'instance pour l'authentification auprès du connecteur 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>

Pour plus d'informations sur les principaux d'instance, reportez-vous à Annonce des principaux d'instance pour Identity and Access Management.

A l'instar des principaux d'instance, Oracle fournit desprincipaux de ressources pour authentifier les ressources qui ne sont pas une instance (comme un bloc-notes jupyter). Chaque ressource possède sa propre identité et est authentifiée à l'aide des certificats qui lui sont ajoutés.

Pour utiliser l'authentification via un principal de ressource avec le connecteur HDFS, indiquez simplement la propriété fs.oci.client.custom.authenticator et définissez la valeur sur com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator.

Etant donné que l'utilisation de principaux de ressource fournit au connecteur un authentificateur personnalisé, il n'est pas nécessaire de configurer les propriétés suivantes :

• 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

L'exemple de code suivant illustre l'utilisation de principaux de ressource pour l'authentification auprès du connecteur 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>

Pour plus d'informations sur les principaux d'instance, reportez-vous à Utilisation de principaux de ressource dans le service Data Science.

Oracle prend en charge l'authentification Kerberos pour la connexion à Object Storage à l'aide du connecteur HDFS.

L'exemple de fichier core-site.xml suivant illustre l'utilisation de Kerberos avec l'authentification par jeton SPNEGO avec le connecteur 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>

Pour plus d'informations sur Kerberos, reportez-vous au Tutoriel sur le protocole Kerberos.

Pour plus d'informations sur SPNEGO, reportez-vous à la norme RFC 4178.

A partir de la version 3.3.0.7.0.0, HDFS prend en charge l'utilisation par défaut du client Apache pour effectuer des appels de service OCI. En effet, le connecteur HDFS s'appuie sur le kit SDK pour Java afin d'envoyer des demandes au serveur. Le kit SDK pour Java prend en charge l'utilisation du connecteur Jersey ApacheConnectorProvider par défaut au lieu du connecteur Jersey HttpUrlConnectorProvider pour permettre au client Apache HttpClient d'effectuer des appels de service OCI.

Pour revenir à l'ancien client Jersey par défaut, définissez la propriété fs.oci.client.jersey.default.connector.enabled dans le fichier core-site.xml sur true. Par défaut, cette valeur est définie sur false.

Vous pouvez définir le nombre maximal de connexions dans le pool de connexions du connecteur HDFS.

Pour ce faire, remplacez la propriété fs.oci.client.apache.max.connection.pool.size dans le fichier core-site.xml par un entier positif qui indique le nombre de connexions à regrouper dans le pool.

Vous pouvez définir les propriétés facultatives suivantes dans le fichier core-site.xml pour configurer un proxy HTTP :

Les objets volumineux sont téléchargés vers Object Storage via des téléchargements multipart. Le fichier est divisé en petites parties qui sont téléchargées en parallèle, ce qui réduit le temps de téléchargement. Cela permet également au connecteur HDFS de réessayer de télécharger les parties ayant échoué au lieu de faire échouer l'ensemble du téléchargement. Toutefois, les téléchargements peuvent échouer de façon temporaire, auquel cas le connecteur tente d'abandonner les fichiers partiellement téléchargés. Etant donné que ces fichiers s'accumulent (et que vous serez facturé pour le stockage), répertoriez les téléchargements régulièrement, puis après un certain nombre de jours, abandonnez-les manuellement en utilisant le kit SDK pour Java.

Pour plus d'informations sur l'utilisation de l'API Object Storage concernant la gestion des téléchargements multipart, reportez-vous à Utilisation des téléchargements multipart vers le serveur.

Les sections suivantes présentent les meilleures pratiques pour l'optimisation des performances et de l'utilisation.

Les fichiers et les systèmes de fichiers HDFS sont référencés via des URI. Le modèle précise le type de système de fichiers, et la partie restante de l'URI est libre pour que l'implémentation du système de fichiers l'interprète comme elle le souhaite.

Object Storage étant une banque d'objets, sa capacité à nommer les objets comme s'ils étaient des fichiers dans un système à fichiers est utilisée pour imiter un système à fichiers réel.

oci://MyBucket@MyNamespace/
		

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

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

La journalisation dans le connecteur est effectuée via SLF4J. SLF4J est une abstraction de journalisation qui permet d'utiliser une bibliothèque de journalisation fournie par l'utilisateur (par exemple, log4j). Pour plus d'informations, reportez-vous au manuel de SLF4J.

L'exemple suivant montre comment activer la journalisation de base sur une sortie standard.

1. Téléchargez le fichier JAR de liaison simple SLF4J : liaison simple SLF4J.
2. Ajoutez le fichier JAR à votre variable d'environnement CLASSPATH.
3. Ajoutez l'argument de machine virtuelle suivant pour activer la journalisation de niveau débogage (par défaut, le niveau information est utilisé) : -Dorg.slf4j.simpleLogger.defaultLogLevel=debug.

Vous pouvez configurer des options de journalisation plus avancées à l'aide de la liaison log4j.

Le connecteur HDFS pour Object Storage inclut une structure de surveillance qui fournit des mesures sur les opérations effectuées à l'aide du connecteur. La structure de surveillance fournit une interface qui peut être implémentée pour utiliser/écouter les mesures générées par le connecteur. Vous pouvez fournir une implémentation personnalisée de cette interface ou utiliser l'implémentation de télémétrie publique OCI incluse dans cette structure.

Mise en route

Pour commencer à utiliser la structure de surveillance du connecteur HDFS, vous devez définir les propriétés suivantes. Une fois ces propriétés définies pour OCIMonitoring, vous pouvez utiliser la vue Explorateur de mesures dans la console OCI pour observer les mesures émises à partir du connecteur 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>

Propriétés com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin

Si la propriété com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin est activée, les propriétés suivantes sont applicables :

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>

Création de votre propre consommateur pour les mesures

L'extension des classes doit avoir un constructeur avec la même signature que celle de la 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 peut être implémentée de trois manières :

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

Cette section contient des informations de résolution pour le connecteur HDFS.

Vous disposez d'une correction de bug ou d'une nouvelle fonctionnalité que vous souhaitez partager ? Le kit SDK est open source et accepte les demandes d'extraction sur GitHub.

Pour être averti lorsqu'une nouvelle version du connecteur HDFS est disponible, abonnez-vous au flux Atom .

Méthodes d'interaction :

• Problèmes dans GitHub : pour enregistrer des bugs et effectuer des demandes de fonctionnalité
• Stack Overflow : utilisez les balises oracle-cloud-infrastructure et oci-hdfs-connector dans votre publication.
• Section Outils de développement des forums Oracle Cloud
• My Oracle Support