Connecteur HDFS pour le stockage d'objets

Pour utiliser le connecteur HDFS, vous devez avoir :

• Un compte Oracle Cloud Infrastructure.
• Un utilisateur créé dans ce compte, dans un groupe, avec une politique qui accorde les autorisations souhaitées pour tout seau à utiliser. Il peut s'agir d'un utilisateur pour vous-même, une autre personne ou un autre système qui doit appeler l'API. Pour des exemples sur la configuration d'un nouvel utilisateur, d'un groupe, d'un compartiment et d'une politique, voir Ajout d'utilisateurs. Pour une politique de stockage d'objets de base, voir Permettre aux administrateurs du stockage d'objets de gérer des seaux et des objets.
• Java 8
• Une valeur de durée de vie de 60. Pour plus d'informations, voir Configuration de la durée de vie de machine virtuelle Java pour les consultations de nom DNS.

Copiez les fichiers JAR regroupés de la bibliothèque et de la bibliothèque de tierce partie dans chaque noeud de la grappe Hadoop afin qu'ils soient inclus dans la variable 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 liste les propriétés supplémentaires que vous pouvez configurer pour une connexion au service de stockage d'objets.

Définition du point d'extrémité de la région

Vous pouvez utiliser plusieurs méthodes pour définir le point d'extrémité de la région pour le connecteur HDFS :

• Définir la propriété de nom d'hôte dans core-site.xml
• Définir la propriété de code ou d'identificateur de région dans core-site.xml
• Autoriser le client ObjectStorage à récupérer le point d'extrémité au moyen du service de métadonnées d'instance

Configuration des propriétés à l'aide de 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 par 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 de données d'identification des utilisateurs ni à fournir des fichiers PEM pour les services en cours d'exécution sur des instances. Chaque instance possède sa propre identité et s'authentifie à l'aide des certificats que les principaux d'instance lui ajoutent.

Pour utiliser l'authentification fondée sur des principaux d'instance avec le connecteur HDFS, il vous suffit de fournir la propriété fs.oci.client.custom.authenticator et de régler la valeur à com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator.

Comme les principaux d'instance fournissent 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 des principaux d'instance pour l'authentification avec le 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, voir Annonce concernant les principaux d'instance pour la gestion des identités et des accès.

Comme pour les principaux d'instance, Oracle fournit des principaux de ressource pour authentifier les ressources qui ne sont pas des instances (un carnet jupyter, par exemple). Chaque ressource possède sa propre identité et s'authentifie à l'aide des certificats qui y sont ajoutés.

Pour utiliser l'authentification fondée sur des principaux de ressource avec le connecteur HDFS, il vous suffit de fournir la propriété fs.oci.client.custom.authenticator et de régler la valeur à com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator.

Comme l'utilisation des 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 des principaux de ressource pour l'authentification avec le 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 de ressource, voir Utilisation des principaux de ressource dans le service de science des données.

Oracle prend en charge l'authentification Kerberos pour la connexion au service de stockage d'objets à 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, voir le tutoriel sur le protocole Kerberos.

Pour plus d'informations sur SPNEGO, voir le document RFC 4178.

À partir de la version 3.3.0.7.0.0, HDFS prend en charge l'utilisation du client Apache par défaut pour effectuer des appels de service OCI. En effet, le connecteur HDFS utilise la trousse SDK pour Java pour envoyer des demandes au serveur. La trousse SDK pour Java prend en charge l'utilisation du connecteur Jersey par défaut ApacheConnectorProvider au lieu du connecteur Jersey HttpUrlConnectorProvider pour permettre au client HTTP Apache d'effectuer des appels de service OCI.

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

Vous pouvez définir le nombre maximal de connexions dans la réserve de connexions 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 spécifie le nombre de connexions dans la réserve.

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

Les objets volumineux sont chargés dans le stockage d'objets à l'aide de chargements en plusieurs parties. Le fichier est fractionné en parties plus petites qui sont chargées en parallèle, ce qui réduit le temps de chargement. Cela permet également au connecteur HDFS de charger à nouveau uniquement les parties en échec et d'éviter ainsi l'échec de l'ensemble du chargement. Toutefois, il arrive que les chargements échouent de façon temporaire. Dans ces cas, le connecteur tentera d'abandonner les fichiers partiellement chargés. Comme ces fichiers se cumulent (et que vous serez facturé pour le stockage), listez les chargements périodiquement, puis après un certain nombre de jours, abandonnez-les manuellement à l'aide de la trousse SDK pour Java.

Vous pouvez trouver des informations sur l'utilisation de l'API Stockage d'objets pour gérer les chargements en plusieurs parties sous Utilisation des chargements en plusieurs parties.

Les sections qui suivent contiennent les meilleures pratiques pour optimiser l'utilisation et la performance.

Les systèmes de fichiers et les fichiers HDFS sont référencés par des URI. Le schéma de l'URI spécifie le type de système de fichiers. La partie restante de l'URI peut être interprétée librement par la mise en oeuvre du système de fichiers.

Comme le stockage d'objets est un magasin d'objets, sa capacité à nommer les objets comme s'ils étaient des fichiers d'un système de données est utilisée pour reproduire un système de données réel.

oci://MyBucket@MyNamespace/
		

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

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

La journalisation dans le connecteur s'effectue au moyen de SLF4J. SLF4J est une abstraction de journalisation qui permet l'utilisation d'une bibliothèque de journalisation fournie par l'utilisateur (par exemple, log4j). Pour plus d'informations, consultez le manuel SLF4J.

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

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

Vous pouvez configurer des options de journalisation plus avancées en utilisant la liaison log4j.

Le connecteur HDFS pour le stockage d'objets comprend un cadre 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 mise en oeuvre pour consommer/écouter les mesures générées par le connecteur. Vous pouvez fournir une mise en oeuvre personnalisée de cette interface ou utiliser la mise en oeuvre de télémétrie publique OCI incluse dans ce cadre.

Introduction

Pour commencer à utiliser le cadre 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éer votre propre consommateur pour les mesures

Les classes d'extension doivent 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 mise en oeuvre de trois façons :

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 dépannage pour le connecteur HDFS.

Vous avez une solution pour un bogue ou une nouvelle fonctionnalité que vous aimeriez partager? La trousse SDK est une solution à code source libre qui accepte les demandes d'extraction sur GitHub.

Si vous souhaitez recevoir un avis lorsqu'une nouvelle version du connecteur HDFS est publiée, abonnez-vous au flux Atom.

Pour rester en contact :

• Problèmes GitHub : Pour signaler des bogues et soumettre des demandes de fonction
• Stack Overflow : Utilisez les balises oracle-cloud-infrastructure et oci-hdfs-connector dans votre publication.
• Section Outils pour développeurs dans les forums Oracle Cloud
• My Oracle Support