Conector HDFS para almacenamiento de objetos

Para utilizar el conector HDFS, debe tener:

• Una cuenta de Oracle Cloud Infrastructure.
• Un usuario creado en esa cuenta, en un grupo con una política que otorgue los permisos deseados para cualquier cubo que desee utilizar. Puede ser un usuario para sí mismo u otra persona/sistema que necesite llamar a la API. Para obtener un ejemplo de cómo configurar un nuevo usuario, grupo, compartimento y política, consulte Adición de usuarios. Para obtener una política básica de Object Storage, consulte Permitir a los administradores de Object Storage gestionar cubos y objetos.
• Java 8
• Un valor TTL de 60. Para obtener más información, consulte Configuración de TTL de JVM para consultas de nombre de DNS.

Copie los archivos jar agrupados de lib y de third-party/lib en cada nodo del cluster de Hadoop para que se incluyan en la 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>

Puede definir las siguientes propiedades del conector HDFS en el archivo core-site.xml. En la página BmcProperties se muestran las propiedades adicionales que puede configurar para una conexión a Object Storage.

Definición del punto final de región

Hay varios métodos que puede utilizar para definir el punto final de región para el conector HDFS:

• Especificando la propiedad hostname en core-site.xml
• Especificando el código de región o la propiedad de identificador de región en core-site.xml
• Permitiendo al cliente ObjectStorage seleccionar el punto final a través del servicio de metadatos de instancia

Configuración de propiedades con core-site.xml

En este ejemplo se muestra cómo se pueden configurar las propiedades en un archivo core-site.xml (los OCID se acortan para abreviar):

<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 proporciona principales de instancia para que ya no sea necesario configurar credenciales de usuario o proporcionar archivos PEM en servicios que se ejecutan en instancias. Cada una de estas instancias tiene su propia identidad y se autentica mediante certificados agregados a la instancia por principales de instancia.

Para utilizar la autenticación de principales de instancia con el conector HDFS, solo tiene que proporcionar la propiedad fs.oci.client.custom.authenticator y definir el valor en com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator.

Puesto que el uso de principales de instancia proporciona al conector un autenticador personalizado, ya no es necesario configurar las siguientes propiedades:

• 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

El siguiente código de ejemplo ilustra el uso de principales de instancia para la autenticación con el conector 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>

Para obtener más información sobre principales de instancia, consulte la sección sobre el anuncio de principales de instancia para Identity and Access Management.

Al igual que los principales del recurso, Oracle proporciona principales de recurso para autenticar los recursos que no son instancias (como un bloc de nota de Jupyter). Cada recurso tiene su propia identidad y se autentica mediante los certificados que se le agregan.

Para utilizar la autenticación de principales de recurso con el conector HDFS, solo tiene que proporcionar la propiedad fs.oci.client.custom.authenticator y definir el valor en com.oracle.bmc.hdfs.auth.ResourcePrincipalsCustomAuthenticator.

Dado que el uso de entidades de recurso proporciona al conector un autenticador personalizado, ya no es necesario configurar las siguientes propiedades:

• 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

El siguiente código de ejemplo ilustra el uso de principales de recurso para la autenticación con el conector 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>

Para obtener más información sobre los principales de recurso, consulte la sección sobre el uso de principales de recurso en el servicio Data Science.

Oracle admite la autenticación de Kerberos para conectarse con Object Storage mediante el conector HDFS.

El siguiente archivo core-site.xml de ejemplo ilustra el uso de Kerberos con autenticación de token SPNEGO con el conector 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>

Para obtener más información sobre Kerberos, consulte el Tutorial del protocolo Kerberos.

Para obtener más información sobre SPNEGO, consulte RFC 4178.

A partir de la versión 3.3.0.7.0.0, HDFS soporta el uso del cliente Apache por defecto para realizar llamadas de servicio de OCI. Esto se debe a que el conector HDFS depende del SDK para Java para enviar solicitudes al servidor. El SDK para Java soporta el uso de ApacheConnectorProvider de Jersey por defecto en lugar de HttpUrlConnectorProvider de Jersey para permitir que HttpClient de Apache realice llamadas de servicio de OCI.

Para volver al cliente antiguo por defecto de Jersey, defina la propiedad fs.oci.client.jersey.default.connector.enabled del archivo core-site.xml en true. Por defecto, este valor está definido en false.

Puede definir el número máximo de conexiones en el pool de conexiones del conector HDFS.

Para ello, cambie la propiedad fs.oci.client.apache.max.connection.pool.size del archivo core-site.xml a un entero positivo que especifique cuántas conexiones hay al pool.

Puede definir las siguientes propiedades opcionales en el archivo core-site.xml para configurar un proxy HTTP:

Los objetos grandes se cargan en Object Storage mediante cargas de varias partes. El archivo se divide en partes más pequeñas que se cargan en paralelo, lo que reduce los tiempos de carga. Esto también permite que el conector HDFS vuelva a intentar las cargas de las partes con fallos en lugar de que falle la carga completa. Sin embargo, las cargas pueden fallar temporalmente y el conector intentará abortar parcialmente los archivos cargados. Dado que estos archivos se acumulan (y se le cobrará el almacenamiento), enumere las cargas periódicamente y, después de un determinado número de días, abórtelas manualmente con el SDK para Java.

En Uso de cargas de varias partesse puede encontrar información sobre el uso de la API de Object Storage para gestionar las cargas de varias partes.

Las siguientes secciones contienen las mejores prácticas para optimizar el uso y el rendimiento.

Se hace referencia a los sistemas de archivos y archivos de HDFS a través de los URI. El esquema especifica el tipo de sistema de archivos y la parte restante del URI está realmente libre para que la implantación del sistema de archivos se interprete como lo desee.

Dado que Object Storage es un almacén de objeto, su capacidad para nombrar objetos como si fueran archivos en un sistema en archivos se utiliza para imitar un sistema en archivos real.

oci://MyBucket@MyNamespace/
		

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

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

El registro en el conector se realiza mediante SLF4J. SLF4J es una abstracción del registro que permite el uso de una biblioteca de registro proporcionado por el usuario (p. ej., log4j). Para obtener más información, consulte el manual de SLF4J.

En los siguientes ejemplos se muestra cómo activar el registro básico para la salida estándar.

1. Descargue el archivo jar de enlace simple de SLF4J: SLF4J Simple Binding
2. Agregue el archivo jar a la classpath
3. Agregue el siguiente argumento de VM para activar el registro de nivel de depuración (por defecto, se utiliza el nivel de información): -Dorg.slf4j.simpleLogger.defaultLogLevel=debug

Puede configurar opciones de registro más avanzadas mediante el enlace de log4j.

El conector HDFS para Object Storage incluye un marco de supervisión que proporciona métricas sobre las operaciones realizadas mediante el conector. El marco de supervisión proporciona una interfaz que se puede implantar para consumir/escuchar las métricas generadas por el conector. Puede proporcionar una implantación personalizada de esta interfaz o puede utilizar la implantación de telemetría pública de OCI incluida en este marco.

Introducción

Para empezar a utilizar la estructura de supervisión del conector HDFS, deberá definir las siguientes propiedades. Una vez definidas estas propiedades para OCIMonitoring, puede utilizar la vista Metric Explorer en la consola de OCI para observar las métricas emitidas desde el conector 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>

Propiedades com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin

Si la propiedad com.oracle.bmc.hdfs.monitoring.OCIMonitorPlugin está activada, se aplican las siguientes propiedades:

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>

Creación de su propio consumidor para métricas

La ampliación de clases debe tener un constructor con la misma firma que la de la clase 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 clase OCIMetric se puede implantar de tres formas:

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

Esta sección contiene información de solución de problemas del conector HDFS.

¿Ha obtenido una corrección para un bug o una nueva función que desee aportar? SDK es un código abierto y acepta solicitudes de obtención en GitHub.

Si desea recibir una notificación cuando se lance una nueva versión del conector HDFS, suscríbase a la fuente Atom.

Formas de contactar:

• Problemas de GitHub: para presentar bugs y solicitudes de funciones
• Desbordamiento de pila: utilice las etiquetas oracle-cloud-infrastructure y oci-hdfs-connector en su publicación
• Sección de herramientas de desarrollador de los foros de Oracle Cloud
• My Oracle Support