产品解决方案文档与社区权益中心定价云市场合作伙伴支持与服务了解阿里云
备案控制台登录/注册
探索云世界
开发者社区 大数据与机器学习 开源大数据平台 E-MapReduce 文章 正文

阿里云 E-MapReduce ClickHouse 操作指南 04 期 — 数据导入

2021-09-07 479
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
EMR Serverless StarRocks，5000CU*H 48000GB*H
简介： 阿里云 E-MapReduce（简称 EMR ）是运行在阿里云平台上的一种大数据处理的系统解决方案。ClickHouse 作为开源的列式存储数据库，主要用于在线分析处理查询（OLAP），能够使用 SQL 查询实时生成分析数据报告。而阿里云 EMR ClickHouse 则提供了开源 OLAP 分析引擎 ClickHouse 的云上托管服务。
+关注继续查看

阿里云 E-MapReduce（简称EMR）是运行在阿里云平台上的一种大数据处理的系统解决方案。ClickHouse 作为开源的列式存储数据库，主要用于在线分析处理查询（OLAP），能够使用 SQL 查询实时生成分析数据报告。而阿里云 EMR ClickHouse 则提供了开源 OLAP 分析引擎 ClickHouse 的云上托管服务。


本系列文章将从以下几个方面详细介绍 EMR ClickHouse 的操作指南：


EMR ClickHouse 操作指南 —数据导入

lALPD4PvON_W-SnNAm7NA-0_1005_622.png

一、从 Spark 导入数据至 ClickHouse

——介绍如何将 Spark 中的数据导入至 ClickHouse 集群

前提条件

  • 开发工具
    • 本地安装了 Java JDK 8。
    • 本地安装了 Maven 3.x。
    • 本地安装了用于 Java 或 Scala 开发的 IDE，推荐 IntelliJ IDEA，且已配置完成 JDK 和 Maven 环境。


背景信息

关于 Spark 的更多介绍，请参见简介。

代码示例

代码示例如下。

package com.company.packageName

import java.util.Properties
import java.util.concurrent.ThreadLocalRandom

import scala.annotation.tailrec

import com.google.common.collect.ImmutableMap
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{SaveMode, SparkSession}

case class Test(id: Int, key1: String, value1: Boolean, key2: Long, value2: Double)

object CKDataImporter extends Logging {

  private var dbName: String = "default"
  private var tableName: String = ""
  private var ckHost: String = ""
  private var ckPort: String = "8123"
  private var user: String = "default"
  private var password: String = ""
  private var local: Boolean = false

  def main(args: Array[String]): Unit = {
    parse(args.toList)
    checkArguments()

    val jdbcUrl = s"jdbc:clickhouse://$ckHost:$ckPort/$dbName"

    logInfo(s"Use jdbc: $jdbcUrl")
    logInfo(s"Use table: $tableName")

    val spark = getSparkSession

    // generate test data
    val rdd = spark.sparkContext.parallelize(1 to 1000).map(i => {
      val rand = ThreadLocalRandom.current()
      val randString = (0 until rand.nextInt(10, 20))
        .map(_ => rand.nextLong())
        .mkString("")

      Test(i, randString, rand.nextBoolean(), rand.nextLong(), rand.nextGaussian())
    })

    val df = spark.createDataFrame(rdd)

    df.write
      .mode(SaveMode.Append)
      .jdbc(jdbcUrl, tableName, getCKJdbcProperties(user, password))
  }

  private def printUsageAndExit(exitCode: Int = 0): Unit = {
    logError("Usage: java -jar /path/to/CKDataImporter.jar [options]")
    logError("  --dbName      设置ClickHouse数据库的名称，默认为default")
    logError("  --tableName   设置ClickHouse库中表的名称")
    logError("  --ckHost      设置ClickHouse地址")
    logError("  --ckPort      设置ClickHouse端口，默认为8123")
    logError("  --user        设置ClickHouse所使用的用户名")
    logError("  --password    设置ClickHouse用户的密码，默认为空")
    logError("  --local       设置此程序使用Spark Local模式运行")
    System.exit(exitCode)
  }

  @tailrec
  private def parse(args: List[String]): Unit = args match {
    case ("--help" | "-h") :: _ =>
      printUsageAndExit()
    case "--dbName" :: value :: tail =>
      dbName = value
      parse(tail)
    case "--tableName" :: value :: tail =>
      tableName = value
      parse(tail)
    case "--ckHost" :: value :: tail =>
      ckHost = value
      parse(tail)
    case "--ckPort" :: value :: tail =>
      ckPort = value
      parse(tail)
    case "--user" :: value :: tail =>
      user = value
      parse(tail)
    case "--password" :: value :: tail =>
      password = value
      parse(tail)
    case "--local" :: tail =>
      local = true
      parse(tail)
    case Nil =>
    case _ =>
      printUsageAndExit(1)
  }

  private def checkArguments(): Unit = {
    if ("".equals(tableName) || "".equals(ckHost)) {
      printUsageAndExit(2)
    }
  }

  private def getCKJdbcProperties(
      user: String,
      password: String,
      batchSize: String = "1000",
      socketTimeout: String = "300000",
      numPartitions: String = "8",
      rewriteBatchedStatements: String = "true"): Properties = {
    val kvMap = ImmutableMap.builder()
      .put("driver", "ru.yandex.clickhouse.ClickHouseDriver")
      .put("user", user)
      .put("password", password)
      .put("batchsize", batchSize)
      .put("socket_timeout", socketTimeout)
      .put("numPartitions", numPartitions)
      .put("rewriteBatchedStatements", rewriteBatchedStatements)
      .build()
    val properties = new Properties
    properties.putAll(kvMap)
    properties
  }

  private def getSparkSession: SparkSession = {
    val builder = SparkSession.builder()
    if (local) {
      builder.master("local[*]")
    }
    builder.appName("ClickHouse-Data-Importer")
    builder.getOrCreate()
  }
}

操作流程

步骤一：创建 ClickHouse 表

  1. 使用 SSH 方式登录 ClickHouse 集群，详情请参见使用SSH连接主节点。
  2. 执行如下命令，进入 ClickHouse 客户端。
clickhouse-client
  1. 创建 ClickHouse 信息。

i. 执行如下命令，创建数据库 clickhouse_database_name。

CREATE DATABASE clickhouse_database_name ON CLUSTER cluster_emr;

阿里云 EMR 会为 ClickHouse 集群自动生成一个名为 cluster_emr 的集群。数据库名您可以自定义。


ii. 执行如下命令，创建表 clickhouse_table_name_local。

CREATE TABLE clickhouse_database_name.clickhouse_table_name_local ON CLUSTER cluster_emr (
  id            UInt32,
  key1            String,
  value1        UInt8,
  key2            Int64,
  value2        Float64
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/clickhouse_database_name/clickhouse_table_name_local', '{replica}')
ORDER BY id


说明：表名您可以自定义，但请确保表名是以 _local 结尾。layer、shard 和 replica 是阿里云 EMR 为 ClickHouse 集群自动生成的宏定义，可以直接使用。


iii. 执行如下命令，创建与表 clickhouse_table_name_local 字段定义一致的表 clickhouse_table_name_all。

说明： 表名您可以自定义，但请确保表名是以 _all 结尾。

CREATE TABLE clickhouse_database_name.clickhouse_table_name_all ON CLUSTER cluster_emr (
  id                    UInt32,
  key1                  String,
  value1                UInt8,
  key2                  Int64,
  value2                Float64
) ENGINE = Distributed(cluster_emr, clickhouse_database_name, clickhouse_table_name_local, rand());

步骤二：编译并打包

  1. 下载并解压CKDataImporter示例到本地。
  2. 在 CMD 命令行中，进入到下载文件中 pom.xml 所在的目录下，执行如下命令打包文件。
mvn clean package

根据您pom.xml文件中artifactId的信息，下载文件中的target目录下会出现CKDataImporter-1.0.0.jar的JAR包。


步骤三：提交作业

  1. 使用 SSH 方式登录 Spark 集群，详情请参见使用SSH连接主节点。
  2. 执行如下命令提交作业。
spark-submit --master yarn \
             --class com.aliyun.emr.CKDataImporter \
             /CKDataImporter-1.0.0.jar \
             --dbName clickhouse_database_name \
             --tableName clickhouse_table_name_all \
             --ckHost ${clickhouse_host}
参数说明
dbNameClickHouse集群数据库的名称，默认为default。本文示例为clickhouse_database_name。
tableNameClickHouse集群数据库中表的名称。本文示例为clickhouse_table_name_all。
ckHostClickHouse集群的Master节点的内网IP地址或公网IP地址。ip地址获取方式，请参见获取主节点的IP地址。

获取主节点的 IP 地址

  1. 进入详情页面。
    1. 登录阿里云E-MapReduce控制台。
    2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
    3. 单击上方的集群管理页签。
    4. 在集群管理页面，单击相应集群所在行的详情。
  1. 在集群基础信息页面的主机信息区域，获取主节点的内网或公网IP地址。

image.png

二、从 Flink 导入数据至 ClickHouse

——介绍如何将 Flink 中的数据导入至 ClickHouse 集群

前提条件

  • 开发工具
    • 本地安装了 Java JDK 8。
    • 本地安装了 Maven 3.x。
    • 本地安装了用于 Java 或 Scala 开发的 IDE，推荐 IntelliJ IDEA，且已配置完成 JDK 和 Maven 环境。


背景信息

关于 Flink 的更多介绍，请参见Apache Flink。

代码示例

代码示例如下。

  • 流处理
package com.company.packageName

import java.util.concurrent.ThreadLocalRandom

import scala.annotation.tailrec

import org.apache.flink.api.common.typeinfo.Types
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala.{StreamTableEnvironment, table2RowDataStream}

object StreamingJob {

  case class Test(id: Int, key1: String, value1: Boolean, key2: Long, value2: Double)

  private var dbName: String = "default"
  private var tableName: String = ""
  private var ckHost: String = ""
  private var ckPort: String = "8123"
  private var user: String = "default"
  private var password: String = ""

  def main(args: Array[String]) {
    parse(args.toList)
    checkArguments()

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    val insertIntoCkSql =
      s"""
        | INSERT INTO $tableName (
        |   id, key1, value1, key2, value2
        | ) VALUES (
        |   ?, ?, ?, ?, ?
        | )
        |""".stripMargin

    val jdbcUrl = s"jdbc:clickhouse://$ckHost:$ckPort/$dbName"

    println(s"jdbc url: $jdbcUrl")
    println(s"insert sql: $insertIntoCkSql")

    val sink = JDBCAppendTableSink
      .builder()
      .setDrivername("ru.yandex.clickhouse.ClickHouseDriver")
      .setDBUrl(jdbcUrl)
      .setUsername(user)
      .setPassword(password)
      .setQuery(insertIntoCkSql)
      .setBatchSize(1000)
      .setParameterTypes(Types.INT, Types.STRING, Types.BOOLEAN, Types.LONG, Types.DOUBLE)
      .build()

    val data: DataStream[Test] = env.fromCollection(1 to 1000).map(i => {
      val rand = ThreadLocalRandom.current()
      val randString = (0 until rand.nextInt(10, 20))
        .map(_ => rand.nextLong())
        .mkString("")
      Test(i, randString, rand.nextBoolean(), rand.nextLong(), rand.nextGaussian())
    })

    val table = table2RowDataStream(tableEnv.fromDataStream(data))
    sink.emitDataStream(table.javaStream)

    // execute program
    env.execute("Flink Streaming Scala API Skeleton")
  }

  private def printUsageAndExit(exitCode: Int = 0): Unit = {
    println("Usage: flink run com.company.packageName.StreamingJob /path/to/flink-clickhouse-demo-1.0.0.jar [options]")
    println("  --dbName      设置ClickHouse数据库的名称，默认为default")
    println("  --tableName   设置ClickHouse库中表的名称")
    println("  --ckHost      设置ClickHouse地址")
    println("  --ckPort      设置ClickHouse端口，默认为8123")
    println("  --user        设置ClickHouse所使用的用户名")
    println("  --password    设置ClickHouse用户的密码，默认为空")
    System.exit(exitCode)
  }

  @tailrec
  private def parse(args: List[String]): Unit = args match {
    case ("--help" | "-h") :: _ =>
      printUsageAndExit()
    case "--dbName" :: value :: tail =>
      dbName = value
      parse(tail)
    case "--tableName" :: value :: tail =>
      tableName = value
      parse(tail)
    case "--ckHost" :: value :: tail =>
      ckHost = value
      parse(tail)
    case "--ckPort" :: value :: tail =>
      ckPort = value
      parse(tail)
    case "--user" :: value :: tail =>
      user = value
      parse(tail)
    case "--password" :: value :: tail =>
      password = value
      parse(tail)
    case Nil =>
    case _ =>
      printUsageAndExit(1)
  }

  private def checkArguments(): Unit = {
    if ("".equals(tableName) || "".equals(ckHost)) {
      printUsageAndExit(2)
    }
  }
}
  • 批处理
package com.company.packageName

import java.util.concurrent.ThreadLocalRandom

import scala.annotation.tailrec

import org.apache.flink.Utils
import org.apache.flink.api.common.typeinfo.Types
import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala.{BatchTableEnvironment, table2RowDataSet}

object BatchJob {

  case class Test(id: Int, key1: String, value1: Boolean, key2: Long, value2: Double)

  private var dbName: String = "default"
  private var tableName: String = ""
  private var ckHost: String = ""
  private var ckPort: String = "8123"
  private var user: String = "default"
  private var password: String = ""

  def main(args: Array[String]) {
    parse(args.toList)
    checkArguments()

    // set up the batch execution environment
    val env = ExecutionEnvironment.getExecutionEnvironment
    val tableEnv = BatchTableEnvironment.create(env)

    val insertIntoCkSql =
      s"""
        | INSERT INTO $tableName (
        |   id, key1, value1, key2, value2
        | ) VALUES (
        |   ?, ?, ?, ?, ?
        | )
        |""".stripMargin
    val jdbcUrl = s"jdbc:clickhouse://$ckHost:$ckPort/$dbName"

    println(s"jdbc url: $jdbcUrl")
    println(s"insert sql: $insertIntoCkSql")

    val sink = JDBCAppendTableSink
      .builder()
      .setDrivername("ru.yandex.clickhouse.ClickHouseDriver")
      .setDBUrl(jdbcUrl)
      .setUsername(user)
      .setPassword(password)
      .setQuery(insertIntoCkSql)
      .setBatchSize(1000)
      .setParameterTypes(Types.INT, Types.STRING, Types.BOOLEAN, Types.LONG, Types.DOUBLE)
      .build()

    val data = env.fromCollection(1 to 1000).map(i => {
      val rand = ThreadLocalRandom.current()
      val randString = (0 until rand.nextInt(10, 20))
        .map(_ => rand.nextLong())
        .mkString("")
      Test(i, randString, rand.nextBoolean(), rand.nextLong(), rand.nextGaussian())
    })

    val table = table2RowDataSet(tableEnv.fromDataSet(data))

    sink.emitDataSet(Utils.convertScalaDatasetToJavaDataset(table))

    // execute program
    env.execute("Flink Batch Scala API Skeleton")
  }

  private def printUsageAndExit(exitCode: Int = 0): Unit = {
    println("Usage: flink run com.company.packageName.StreamingJob /path/to/flink-clickhouse-demo-1.0.0.jar [options]")
    println("  --dbName      设置ClickHouse数据库的名称，默认为default")
    println("  --tableName   设置ClickHouse库中表的名称")
    println("  --ckHost      设置ClickHouse地址")
    println("  --ckPort      设置ClickHouse端口，默认为8123")
    println("  --user        设置ClickHouse所使用的用户名")
    println("  --password    设置ClickHouse用户的密码，默认为空")
    System.exit(exitCode)
  }

  @tailrec
  private def parse(args: List[String]): Unit = args match {
    case ("--help" | "-h") :: _ =>
      printUsageAndExit()
    case "--dbName" :: value :: tail =>
      dbName = value
      parse(tail)
    case "--tableName" :: value :: tail =>
      tableName = value
      parse(tail)
    case "--ckHost" :: value :: tail =>
      ckHost = value
      parse(tail)
    case "--ckPort" :: value :: tail =>
      ckPort = value
      parse(tail)
    case "--user" :: value :: tail =>
      user = value
      parse(tail)
    case "--password" :: value :: tail =>
      password = value
      parse(tail)
    case Nil =>
    case _ =>
      printUsageAndExit(1)
  }

  private def checkArguments(): Unit = {
    if ("".equals(tableName) || "".equals(ckHost)) {
      printUsageAndExit(2)
    }
  }
}

操作流程

步骤一：创建 ClickHouse 表

  1. 使用 SSH 方式登录 ClickHouse 集群，详情请参见使用SSH连接主节点。
  2. 执行如下命令，进入 ClickHouse 客户端。
clickhouse-client
  1. 创建 ClickHouse 信息。

i. 执行如下命令，创建数据库 clickhouse_database_name。

CREATE DATABASE clickhouse_database_name ON CLUSTER cluster_emr;

阿里云 EMR 会为 ClickHouse 集群自动生成一个名为 cluster_emr 的集群。数据库名您可以自定义。


ii. 执行如下命令，创建表 clickhouse_table_name_local。

CREATE TABLE clickhouse_database_name.clickhouse_table_name_local ON CLUSTER cluster_emr (
  id            UInt32,
  key1            String,
  value1        UInt8,
  key2            Int64,
  value2        Float64
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/clickhouse_database_name/clickhouse_table_name_local', '{replica}')
ORDER BY id


说明：表名您可以自定义，但请确保表名是以_local结尾。layer、shard 和 replica 是阿里云 EMR 为 ClickHouse 集群自动生成的宏定义，可以直接使用。


iii. 执行如下命令，创建与表 clickhouse_table_name_local 字段定义一致的表clickhouse_table_name_all。

说明： 表名您可以自定义，但请确保表名是以_all结尾。

CREATE TABLE clickhouse_database_name.clickhouse_table_name_all ON CLUSTER cluster_emr (
  id                    UInt32,
  key1                  String,
  value1                UInt8,
  key2                  Int64,
  value2                Float64
) ENGINE = Distributed(cluster_emr, clickhouse_database_name, clickhouse_table_name_local, rand());

步骤二：编译并打包

  1. 下载并解压flink-clickhouse-demo.tgz示例到本地。
  2. 在 CMD 命令行中，进入到下载文件中 pom.xml 所在的目录下，执行如下命令打包文件。
mvn clean package

根据您 pom.xml 文件中 artifactId 的信息，下载文件中的 target 目录下会出现 flink-clickhouse-demo-1.0.0.jar 的JAR包。


步骤三：提交作业

  1. 使用 SSH 方式登录 Flink 集群，详情请参见使用SSH连接主节点。
  2. 执行如下命令提交作业。代码示例如下：
    • 流作业
flink run -m yarn-cluster \
          -c com.aliyun.emr.StreamingJob \
          flink-clickhouse-demo-1.0.0.jar \
          --dbName clickhouse_database_name \
          --tableName clickhouse_table_name_all \
          --ckHost ${clickhouse_host}
    • 批作业
flink run -m yarn-cluster \
          -c com.aliyun.emr.BatchJob \
          flink-clickhouse-demo-1.0.0.jar \
          --dbName clickhouse_database_name \
          --tableName clickhouse_table_name_all \
          --ckHost ${clickhouse_host}
参数说明
dbNameClickHouse集群数据库的名称，默认为default。本文示例为clickhouse_database_name。
tableNameClickHouse集群数据库中表的名称。本文示例为clickhouse_table_name_all。
ckHostClickHouse集群的Master节点的内网IP地址或公网IP地址。ip地址获取方式，请参见获取主节点的IP地址。

获取主节点的 IP 地址

  1. 进入详情页面。
    1. 登录阿里云E-MapReduce控制台。
    2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
    3. 单击上方的集群管理页签。
    4. 在集群管理页面，单击相应集群所在行的详情。
  1. 在集群基础信息页面的主机信息区域，获取主节点的内网或公网IP地址。

image.png


三、从 HDFS 导入数据至 ClickHouse

——可以通过 HDFS 表引擎或表函数读写数据

前提条件


使用 HDFS 表引擎读写数据

语法

CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
  name1 [type1],
  name2 [type2],
  ...
)
Engine = HDFS(uri, format)

其中，涉及参数描述如下表所示。

参数描述
db数据库名。
table_name表名。
name1/name2列名。
tyep1/type2列的类型。
uriHDFS上文件的地址。


说明

  • 不可以是目录地址。
  • 文件所属的目录需要存在，如果不存在，则写数据时会报错。
format文件的类型。

示例

  1. 创建业务表和 HDFS 表

i. 使用 SSH 方式登录 ClickHouse 集群，详情请参见登录集群。

ii. 执行如下命令，进入 ClickHouse 客户端。

clickhouse-client

iii. 执行以下命令，创建数据库 hdfs。

CREATE DATABASE IF NOT EXISTS hdfs ON CLUSTER cluster_emr;

iv. 执行以下命令，创建表 orders。

CREATE TABLE IF NOT EXISTS hdfs.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
ENGINE = HDFS('hdfs://192.168.**.**:9000/orders.csv', 'CSV');


说明 本文示例是将示例数据上传到了HDFS集群的根目录下。代码中的192.168.**.**为HDFS集群的emr-header-1节点的内网IP地址，您可以在EMR控制台集群管理页签中的主机列表页面查看。

v. 执行以下命令，创建数据库 product。

CREATE DATABASE IF NOT EXISTS product ON CLUSTER cluster_emr;

vi. 执行以下命令，创建业务表 orders。

CREATE TABLE IF NOT EXISTS product.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = ReplicatedMergeTree('/cluster_emr/product/orders/{shard}', '{replica}')
PARTITION BY toYYYYMMDD(date)
ORDER BY toYYYYMMDD(date);

vii. 执行以下命令，创建业务表 orders_all。

CREATE TABLE IF NOT EXISTS product.orders_all ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = Distributed(cluster_emr, product, orders, rand());


  1. 使用 HDFS 表引擎导入数据

i. 下载并上传示例数据orders.csv至 HDFS 集群的目录下。
说明 本文示例上传到了 HDFS 集群的根目录下。

ii. 执行以下命令，导入数据。

INSERT INTO product.orders_all
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
  hdfs.orders;

iii. 执行以下命令，检查数据一致性。

SELECT
  a.*
FROM
  hdfs.orders a
LEFT ANTI JOIN
  product.orders_all
USING uid;
  1. 使用 HDFS 表引擎导出数据

i. 执行以下命令，构造数据。

INSERT INTO product.orders_all VALUES \
  (60333391,'2021-08-04 11:26:01',49358700,89) \
  (38826285,'2021-08-03 10:47:29',25166907,27) \
  (10793515,'2021-07-31 02:10:31',95584454,68) \
  (70246093,'2021-08-01 00:00:08',82355887,97) \
  (70149691,'2021-08-02 12:35:45',68748652,1)  \
  (87307646,'2021-08-03 19:45:23',16898681,71) \
  (61694574,'2021-08-04 23:23:32',79494853,35) \
  (61337789,'2021-08-02 07:10:42',23792355,55) \
  (66879038,'2021-08-01 16:13:19',95820038,89);

ii. 执行以下命令，导出数据。

INSERT INTO hdfs.orders
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
 product.orders_all;

iii. 执行以下命令，可以检查数据一致性。

SELECT
  a.*
FROM
  hdfs.orders
RIGHT ANTI JOIN
  product.orders_all a
USING uid;


使用 HDFS 表函数读写数据

语法

hdfs(uri, format, structure);

其中，涉及参数描述如下表所示。

参数描述
uriHDFS上文件的地址。

说明

  • 不可以是目录地址。
  • 文件所属的目录需要存在，如果不存在，则写数据时会报错。
format文件的类型。
structure表中字段的类型。例如，column1 UInt32，column2 String。

示例

  1. 创建数据库和业务表

i. 使用 SSH 方式登录 ClickHouse 集群，详情请参见登录集群。

ii. 执行如下命令，进入 ClickHouse 客户端。

clickhouse-client

iii. 执行以下命令，创建数据库 product。

CREATE DATABASE IF NOT EXISTS product ON CLUSTER cluster_emr;

iv. 执行以下命令，创建业务表 orders。

CREATE TABLE IF NOT EXISTS product.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = ReplicatedMergeTree('/cluster_emr/product/orders/{shard}', '{replica}')
PARTITION BY toYYYYMMDD(date)
ORDER BY toYYYYMMDD(date);

v. 执行以下命令，创建业务表 orders_all。

CREATE TABLE IF NOT EXISTS product.orders_all ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = Distributed(cluster_emr, product, orders, rand());
  1. 使用 HDFS 表引擎导入数据

i. 下载并上传示例数据orders.csv至HDFS集群的目录下。

说明 本文示例上传到了 HDFS 集群的根目录下。

ii. 执行以下命令，导入数据。

INSERT INTO product.orders_all
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
  hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV', 'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32');

iii. 执行以下命令，可以检查数据一致性。

SELECT
  a.*
FROM
  hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV', 'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32') a
LEFT ANTI JOIN
  product.orders_all
USING uid;
  1. 使用 HDFS 表引擎导出数据

i. 执行以下命令，构造数据。

INSERT INTO product.orders_all VALUES \
  (60333391,'2021-08-04 11:26:01',49358700,89) \
  (38826285,'2021-08-03 10:47:29',25166907,27) \
  (10793515,'2021-07-31 02:10:31',95584454,68) \
  (70246093,'2021-08-01 00:00:08',82355887,97) \
  (70149691,'2021-08-02 12:35:45',68748652,1)  \
  (87307646,'2021-08-03 19:45:23',16898681,71) \
  (61694574,'2021-08-04 23:23:32',79494853,35) \
  (61337789,'2021-08-02 07:10:42',23792355,55) \
  (66879038,'2021-08-01 16:13:19',95820038,89);

ii. 执行以下命令，导出数据。

INSERT INTO FUNCTION
  hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV', 'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
  product.orders_all;

iii. 执行以下命令，可以检查数据一致性。

SELECT
  a.*
FROM
  hdfs('hdfs://192.168.**.**:9000/orders.csv', 'CSV', 'uid UInt32, date DateTime, skuId UInt32, order_revenue UInt32')
RIGHT ANTI JOIN
  product.orders_all a

配置

EMR ClickHouse 允许使用对 HDFS 进行配置：

  • 全局生效的 HDFS 配置。
<hdfs>
  <dfs_default_replica>3</dfs_default_replica>
</hdfs>


说明 查询参数时将下划线（_）替换为半角句号（.）即可。例如，您要查询EMR中的参数dfs_default_replica，则可以在官网文档中搜索dfs.default.replica。


  • 仅对${user}用户生效的 HDFS 配置，用户配置与全局配置相同的键不同值时，会覆盖全局配置
<hdfs_${user}>
  <dfs_default_replica>3</dfs_default_replica>
</hdfs_${user}>

四、从 RDS 导入数据至 ClickHouse 集群

——可以通过 RDS MySQL 表引擎或表函数导入数据至 ClickHouse 集群

前提条件


使用 RDS MySQL 表引擎导入数据

语法

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1] [TTL expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2] [TTL expr2],
    ...
) ENGINE = MySQL('host:port', 'database', 'table', 'user', 'password'[, replace_query, 'on_duplicate_clause']);

其中，涉及参数描述如下表所示。

参数描述
db数据库名。
table_name表名。
cluster集群标识。
name1/name2列名。
tyep1/type2列的类型。
host:portRDS MySQL的地址，可以在RDS MySQL管理控制台中数据库连接中进行查看。
databaseRDS MySQL中的数据库名。
tableRDS MySQL中的表名。
user用户名，该用户具有访问上述RDS MySQL中库中的表的权限。
passworduser对应的密码。
replace_query是否将INSERT INTO查询转换为REPLACE INTO的标志。设置为1，表示替换查询。
on_duplicate_clause会被添加到INSERT语句中。例如，INSERT INTO t (c1,c2) VALUES ('a', 2) ON DUPLICATE KEY UPDATE c2 = c2 + 1，此时需要指定on_duplicate_clause为UPDATE c2 = c2 + 1。

示例

  1. 在 RDS MySQL 实例中，创建原始数据表并导入原始数据。

i. 连接 MySQL 实例，详情请参见通过客户端、命令行连接RDS MySQL。

ii. 执行以下命令，创建原始数据表。

CREATE TABLE `origin`.`orders` (
  `uid` int(10) unsigned DEFAULT NULL,
  `date` datetime DEFAULT NULL,
  `skuId` int(10) unsigned DEFAULT NULL,
  `order_revenue` int(10) unsigned DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8

iii. 执行以下命令，导入原始数据。

INSERT INTO `origin`.`orders` VALUES(60333391, '2021-08-04 11:26:01', 49358700, 89),
       (38826285, '2021-08-03 10:47:29', 25166907, 27),
       (10793515, '2021-07-31 02:10:31', 95584454, 68),
       (70246093, '2021-08-01 00:00:08', 82355887, 97),
       (70149691, '2021-08-02 12:35:45', 68748652, 1),
       (87307646, '2021-08-03 19:45:23', 16898681, 71),
       (61694574, '2021-08-04 23:23:32', 79494853, 35),
       (61337789, '2021-08-02 07:10:42', 23792355, 55),
       (66879038, '2021-08-01 16:13:19', 95820038, 89);
  1. 在 ClickHouse 集群中，执行以下操作。

i. 使用 SSH 方式登录 ClickHouse 集群，详情请参见登录集群。

ii. 执行以下命令，进入 ClickHouse 客户端。

clickhouse-client

iii. 执行以下命令，创建数据库 mysql。

CREATE DATABASE IF NOT EXISTS mysql;

iv. 执行以下命令，创建表 orders。

CREATE TABLE mysql.orders
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
ENGINE = MySQL('host:port', 'origin', 'orders', 'user', 'password');

v. 执行以下命令，创建数据库 product。

CREATE DATABASE IF NOT EXISTS product ON CLUSTER cluster_emr;

vi. 执行以下命令，创建业务表 orders。

CREATE TABLE IF NOT EXISTS product.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = ReplicatedMergeTree('/cluster_emr/product/orders/{shard}', '{replica}')
PARTITION BY toYYYYMMDD(date)
ORDER BY toYYYYMMDD(date);

vii. 执行以下命令，创建业务表orders_all。

CREATE TABLE IF NOT EXISTS product.orders_all ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = Distributed(cluster_emr, product, orders, rand());

viii. 执行以下命令，导入数据。

INSERT INTO product.orders_all
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
  mysql.orders;

ix. 执行以下命令，查询数据。

SELECT a.*
FROM mysql.orders AS a
ANTI LEFT JOIN product.orders_all USING (uid);


说明 查询数据为空时正常。


使用 RDS MySQL 表函数导入数据

语法

mysql('host:port', 'database', 'table', 'user', 'password'[, replace_query, 'on_duplicate_clause'])

其中，涉及参数描述如下表所示。

参数描述
host:portRDS MySQL的地址，您可以在RDS MySQL管理控制台中的数据库连接中查看。
databaseRDS MySQL中的数据库名。
tableRDS MySQL中的表名。
user用户名，该用户具有访问上述RDS MySQL中库中的表的权限。
passworduser对应的密码。
replace_query是否将INSERT INTO查询转换为REPLACE INTO的标志。设置为1，表示替换查询。
on_duplicate_clause会被添加到INSERT语句中。例如，INSERT INTO t (c1,c2) VALUES ('a', 2) ON DUPLICATE KEY UPDATE c2 = c2 + 1，此时需要指定on_duplicate_clause为UPDATE c2 = c2 + 1。

示例

  1. 在 RDS MySQL 实例中，创建表并插入数据。

i. 连接 MySQL 实例，详情请参见通过客户端、命令行连接RDS MySQL。

ii. 执行以下命令，创建表 orders。

CREATE TABLE `origin`.`orders` (
  `uid` int(10) unsigned DEFAULT NULL,
  `date` datetime DEFAULT NULL,
  `skuId` int(10) unsigned DEFAULT NULL,
  `order_revenue` int(10) unsigned DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8

iii. 执行以下命令，插入数据。

INSERT INTO `origin`.`orders` VALUES(60333391, '2021-08-04 11:26:01', 49358700, 89),
       (38826285, '2021-08-03 10:47:29', 25166907, 27),
       (10793515, '2021-07-31 02:10:31', 95584454, 68),
       (70246093, '2021-08-01 00:00:08', 82355887, 97),
       (70149691, '2021-08-02 12:35:45', 68748652, 1),
       (87307646, '2021-08-03 19:45:23', 16898681, 71),
       (61694574, '2021-08-04 23:23:32', 79494853, 35),
       (61337789, '2021-08-02 07:10:42', 23792355, 55),
       (66879038, '2021-08-01 16:13:19', 95820038, 89) ;
  1. 在 ClickHouse 集群中，执行以下操作。

i. 使用 SSH 方式登录 ClickHouse 集群，详情请参见登录集群。

ii. 执行如下命令，进入 ClickHouse 客户端。

clickhouse-client

iii. 执行以下命令，创建数据库 product。

CREATE DATABASE IF NOT EXISTS product ON CLUSTER cluster_emr;

iv.执行以下命令，创建表 orders。

CREATE TABLE IF NOT EXISTS product.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = ReplicatedMergeTree('/cluster_emr/product/orders/{shard}', '{replica}')
PARTITION BY toYYYYMMDD(date)
ORDER BY toYYYYMMDD(date);

v. 执行以下命令，创建表 orders_all。

CREATE TABLE IF NOT EXISTS product.orders_all ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = Distributed(cluster_emr, product, orders, rand());

vi. 执行以下命令，导入数据。

INSERT INTO product.orders_all
SELECT
  uid,
  date,
  skuId,
  order_revenue
FROM
  mysql('host:port', 'origin', 'orders', 'user', 'password');

vii. 执行以下命令，查询数据。

SELECT a.*
FROM
  mysql('host:port', 'origin', 'orders', 'user', 'password') AS a
ANTI LEFT JOIN product.orders_all USING (uid);


说明 查询数据为空时正常。

如果您需要导出数据，则将业务表数据写入 MySQL 表函数即可。写入命令如下。

INSERT INTO FUNCTION
  mysql('host:port', 'origin', 'orders', 'user', 'password')
FROM
  product.orders_all;

五、从 Kafka 导入数据至 ClickHouse

——可以通过 Kafka 表引擎导入数据至 ClickHouse 集群

前提条件

使用限制

Kafka集群和ClickHouse集群需要在同一VPC下。

语法

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'host1:port1,host2:port2',
    kafka_topic_list = 'topic1,topic2,...',
    kafka_group_name = 'group_name',
    kafka_format = 'data_format';

其中，涉及参数描述如下表所示。

参数描述
db数据库名。
table_name表名。
cluster集群标识。
name1/name2列名。
tyep1/type2列的类型。
kafka_broker_listKafka Broker的地址及端口。

Kafka集群所有节点的内网IP地址及端口，您可以在EMR控制台集群管理页签中的主机列表页面查看。

kafka_topic_list订阅的Topic名称。
kafka_group_nameKafka consumer的分组名称。
kafka_format数据的类型。例如，CSV和JSONEachRow等，详细信息请参见Formats for Input and Output Data。

示例

  1. 在 ClickHouse 集群中执行以下操作。

i. 使用SSH方式登录 ClickHouse 集群，详情请参见登录集群。

ii. 执行如下命令，进入 ClickHouse 客户端。

clickhouse-client

iii. 执行如下命令，创建数据库 kafka。

CREATE DATABASE IF NOT EXISTS kafka ON CLUSTER cluster_emr;


说明 数据库名您可以自定义。本文示例中的cluster_emr是集群默认的标识，如果您修改过，请填写正确的集群标识，您也可以在EMR控制台ClickHouse服务的配置页面，在搜索区域搜索clickhouse_remote_servers参数查看。

iv. 执行如下命令，创建 Kafka 表。

CREATE TABLE IF NOT EXISTS kafka.consumer ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
ENGINE = Kafka()
SETTINGS
  kafka_broker_list = '192.168.**.**:9092,192.168.**.**:9092,192.168.**.**:9092',
  kafka_topic_list = 'clickhouse_test',
  kafka_group_name = 'clickhouse_test',
  kafka_format = 'CSV';

kafka_broker_list为 Kafka 集群所有节点的内网 IP 地址及端口，您可以在 EMR 控制台集群管理页签中的主机列表页面查看。其余参数含义请参见语法。

image.png

v. 执行如下命令，创建数据库 product。

CREATE DATABASE IF NOT EXISTS product ON CLUSTER cluster_emr;

vi. 执行以下命令，创建本地表。

CREATE TABLE IF NOT EXISTS product.orders ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = ReplicatedMergeTree('/cluster_emr/product/orders/{shard}', '{replica}')
PARTITION BY toYYYYMMDD(date)
ORDER BY toYYYYMMDD(date);

vii. 执行以下命令，创建分布式表。

CREATE TABLE IF NOT EXISTS product.orders_all ON CLUSTER cluster_emr
(
    `uid` UInt32,
    `date` DateTime,
    `skuId` UInt32,
    `order_revenue` UInt32
)
Engine = Distributed(cluster_emr, product, orders, rand());

viii. 执行以下命令，创建 MATERIALIZED VIEW 自动导数据。

CREATE MATERIALIZED VIEW IF NOT EXISTS product.kafka_load ON CLUSTER cluster_emr TO product.orders AS
SELECT *
FROM kafka.consumer;
  1. 在 Kafka 集群中执行以下操作。

i. 使用 SSH 方式登录 Kafka 集群，详情请参见登录集群。

ii. 在 Kafka 集群的命令行窗口，执行如下命令运行 Kafka 的生产者。

/usr/lib/kafka-current/bin/kafka-console-producer.sh --broker-list 192.168.**.**:9092,192.168.**.**:9092,192.168.**.**:9092 --topic clickhouse_test

iii. 执行以下命令，输入测试数据。

38826285,2021-08-03 10:47:29,25166907,27
10793515,2021-07-31 02:10:31,95584454,68
70246093,2021-08-01 00:00:08,82355887,97
70149691,2021-08-02 12:35:45,68748652,1
87307646,2021-08-03 19:45:23,16898681,71
61694574,2021-08-04 23:23:32,79494853,35
61337789,2021-08-02 07:10:42,23792355,55
66879038,2021-08-01 16:13:19,95820038,89

image.png

  1. 在 ClickHouse 命令窗口中，执行以下命令，可以查看从 Kafka 中导入至 ClickHouse 集群的数据。

您可以校验查询到的数据与源数据是否一致。

SELECT * FROM product.orders_all;

image.png


后续

您已经学习了数据导入，本系列还包括其他内容：



获取更详细的信息，点击下方链接查看：

EMR官网：https://www.aliyun.com/product/emapreduce

EMR ClickHouse ：https://help.aliyun.com/document_detail/212195.html


扫描下方二维码加入 EMR 相关产品钉钉交流群一起参与讨论吧！

lALPD26eQMAeAf_NAd7NAvs_763_478.png

文章标签：
开源大数据平台 E-MapReduce
云消息队列 Kafka 版
云数据库 ClickHouse
云数据库 RDS MySQL 版
实时计算 Flink版
Java
消息中间件
流计算
分布式计算
关系型数据库
网络安全
MySQL
Kafka
RDS
数据库
关键词：
阿里云e-mapreduce
云数据库 ClickHouse阿里云
阿里云clickhouse
云数据库 ClickHouse操作
云数据库 ClickHouse e-mapreduce
相关实践学习
EMR数据湖开发治理之用户画像分析
通过本场景，你可以基于E-MapReduce + DLF + OSS-HDFS + DataWorks在云上快速体验完整的数据湖开发治理方案。
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上，利用开源大数据生态系统，包括 Hadoop、Spark、HBase，为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
阿里云E-MapReduce团队
目录
相关文章
庄同学(魏庄)
|
3月前
|
存储 缓存 Cloud Native
阿里云 ClickHouse 企业版首发邀测&云原生 ClickHouse 技术揭秘
云数据库 ClickHouse 企业版是阿里云和 ClickHouse, Inc 战略合作打造的云原生ClickHouse 产品。企业版推出专属 SharedMergeTree 云原生引擎，支持存算分离，Serverless 秒级实时弹性，集群吞吐和查询效率线性扩展及 Lightweight update 实时更新能力。本文将详细揭秘 SharedMergeTree 实现机制，实时弹性扩展实现原理，lightweight update 技术实现原理，同时对企业版和开源版进行详细的性能测试对比。
庄同学(魏庄)
294 0
凤豪
|
3月前
|
存储 缓存 Cloud Native
阿里云 ClickHouse 企业版云原生 ClickHouse 技术揭秘
云数据库 ClickHouse 企业版是阿里云和 ClickHouse, Inc 战略合作打造的云原生ClickHouse 产品。企业版推出专属 SharedMergeTree 云原生引擎，支持存算分离，Serverless 秒级实时弹性，集群吞吐和查询效率线性扩展及 Lightweight update 实时更新能力。本文将详细揭秘 SharedMergeTree 实现机制，实时弹性扩展实现原理，lightweight update 技术实现原理，同时对企业版和开源版进行详细的性能测试对比。
凤豪
1174 1
阿里云 ClickHouse 企业版云原生 ClickHouse 技术揭秘
阿里云瑶池数据库SelectDB
|
3月前
|
SQL 分布式计算 测试技术
从 Clickhouse 到阿里云数据库 SelectDB 版内核 Apache Doris：有赞业务场景下性能测试与迁移验证
从 Clickhouse 到阿里云数据库 SelectDB 版内核 Apache Doris 迁移实践：有赞查询提速近 10 倍，OLAP 分析更实时高效！
阿里云瑶池数据库SelectDB
452 1
从 Clickhouse 到阿里云数据库 SelectDB 版内核 Apache Doris：有赞业务场景下性能测试与迁移验证
开发工程师
|
7月前
|
数据库
《阿里云产品手册2022-2023 版》——云数据库 ClickHouse
《阿里云产品手册2022-2023 版》——云数据库 ClickHouse
开发工程师
90 0
凤豪
|
存储 缓存 运维
阿里云数据库 ClickHouse 云原生版产品解析
本文简要概述了开源 ClickHouse 应用行业场景及性能优势，以及业界对于ClickHouse的关注度趋势和使用情况。 详细说明阿里云数据库ClickHouse 云原生版架构和功能特性，分析自建ClickHouse 用户使用过程中遇到的问题，以及如何通过云原生版本解决自建的部分难点问题。最后介绍云原生版本试用申请流程。
凤豪
3189 1
阿里云数据库 ClickHouse 云原生版产品解析
扬流
|
数据挖掘
初识阿里云E-MapReduce ClickHouse，到底有哪些优势？
本文将从核心能力、主要优势，及典型应用场景为大家继续介绍EMR~
扬流
912 0
初识阿里云E-MapReduce ClickHouse，到底有哪些优势？
云上猫猫
|
分布式计算 Hadoop C++
第4期：初识阿里云E-MapReduce ClickHouse，到底有哪些优势？
本次将从核心能力、主要优势，及典型应用场景为大家继续介绍EMR~
云上猫猫
273 0
第4期：初识阿里云E-MapReduce ClickHouse，到底有哪些优势？
阿里云E-MapReduce团队
|
存储 SQL 运维
阿里云 E-MapReduce ClickHouse 操作指南 05 期 — 常见问题
阿里云 E-MapReduce（简称 EMR ）是运行在阿里云平台上的一种大数据处理的系统解决方案。ClickHouse 作为开源的列式存储数据库，主要用于在线分析处理查询（OLAP），能够使用 SQL 查询实时生成分析数据报告。而阿里云 EMR ClickHouse 则提供了开源 OLAP 分析引擎 ClickHouse 的云上托管服务。
阿里云E-MapReduce团队
560 0
阿里云 E-MapReduce ClickHouse 操作指南 05 期 — 常见问题
阿里云E-MapReduce团队
|
存储 SQL XML
阿里云 E-MapReduce ClickHouse 操作指南 03 期 — ClickHouse 运维
阿里云 E-MapReduce（简称 EMR ）是运行在阿里云平台上的一种大数据处理的系统解决方案。ClickHouse 作为开源的列式存储数据库，主要用于在线分析处理查询（OLAP），能够使用 SQL 查询实时生成分析数据报告。而阿里云 EMR ClickHouse 则提供了开源 OLAP 分析引擎 ClickHouse 的云上托管服务。
阿里云E-MapReduce团队
651 0
阿里云 E-MapReduce ClickHouse 操作指南 03 期 — ClickHouse 运维
阿里云E-MapReduce团队
|
存储 分布式计算 网络协议
阿里云 E-MapReduce ClickHouse 操作指南 02期 — 快速入门
阿里云 E-MapReduce（简称EMR）是运行在阿里云平台上的一种大数据处理的系统解决方案。ClickHouse 作为开源的列式存储数据库，主要用于在线分析处理查询（OLAP），能够使用 SQL 查询实时生成分析数据报告。而阿里云 EMR ClickHouse 则提供了开源 OLAP 分析引擎 ClickHouse 的云上托管服务。
阿里云E-MapReduce团队
552 0
阿里云 E-MapReduce ClickHouse 操作指南 02期 — 快速入门
大数据与机器学习
开源大数据平台 E-MapReduce
热门文章
最新文章
1
如何在E-MapReduce中玩转OSS
2
玩转阿里云EMR三部曲-高级篇 交互式查询及统一数据源
3
2016年北京中国云计算技术大会见闻
4
漫谈分布式计算框架
5
浅谈 Spark 的多语言支持（修订版）
6
#Apache spark系列技术直播# 第六讲【 What's New in Apache Spark 2.4? 】
7
钉钉群直播【基于Spark实现的MLSQL如何帮助企业构建数据中台】
8
Spark Codegen浅析
9
免费公测｜阿里云EMR Serverless StarRocks 公测正式开启！
10
Data Lake 三剑客——Delta、Hudi、Iceberg 对比分析
1
王日宇：基于 StarRocks 和 Paimon 打造湖仓分析新范式
43
2
【2023云栖】陈守元：阿里云开源大数据产品年度发布
71
3
李钰：从 OLAP 到湖仓，阿里云与 StarRocks 的共创共荣
63
4
python爬虫中 HTTP 到 HTTPS 的自动转换
33
5
【云栖2023】李钰：阿里云 E-MapReduce 全面开启 Serverless 时代
133
6
耳朵经济快速增长背后，喜马拉雅数据价值如何释放 | 创新场景
46233
7
来云栖大会看 “ 云+StarRocks 3.0：极速统一的湖仓新范式 ”
245
8
Json实现根据关键词搜索请求淘宝商品列表数据方法，淘宝商品列表数据接口，淘宝API接口申请指南
186
9
杭州 Meetup｜ Apache Kyuubi & Celeborn，助力 Spark 拥抱云原生
669
10
阿里云“玩转云上 StarRocks3.0 湖仓分析”训练营火热报名中，开启数据分析新范式
682
相关产品
开源大数据平台 E-MapReduce
文档详情 产品详情
相关课程
更多
阿里云物联网平台专题介绍 企业上云攻略-阿里云网络产品应用系列教程 阿里云图数据库GDB入门与应用 云原生基础概念及阿里云云原生产品介绍 阿里云负载均衡SLB实战演练 跟阿里云技术专家学习智能推荐系统
相关电子书
更多
ClickHouse在手淘流量分析应用实践Jason Xu 云数据库clickhouse最佳实践 阿里云 ClickHouse 企业版技术白皮书
相关实验场景
更多
使用阿里云ECS搭建WIKI知识库 基于阿里云E-HPC平台的WRF气象预报实验 基于阿里云DeepGPU实例，让AI带你畅玩杭州 基于阿里云DeepGPU实例，用AI画唯美国风少女 基于阿里云E-HPC和OpenFoam进行流体仿真实验 基于Hologres+Flink搭建GitHub实时数据大屏
推荐文章
更多
重磅来袭！参与评测赢Iphone14 pro！ 文件存储NAS评测征集令！ 招募！寻找技术人的伯乐！ 乘风者计划邀您入驻社区，精彩权益即刻享
推荐镜像
更多
clickhouse postgresql mongodb
下一篇
手动搭建WordPress（CentOS 8）
为什么选择阿里云什么是云计算全球基础设施技术领先稳定可靠安全合规分析师报告
产品和定价全部产品免费试用产品动态产品定价价格计算器云上成本管理
解决方案技术解决方案
文档与社区文档开发者社区天池大赛培训与认证
权益中心免费试用高校计划企业扶持计划推荐返现计划
支持与服务基础服务企业增值服务迁云服务官网公告系统状态信任中心
关注阿里云
关注阿里云公众号或下载阿里云APP，关注云资讯，随时随地运维管控云服务
阿里云APP阿里云微信
售前咨询：95187-1
售后服务：400-80-13260
法律声明及隐私权政策Cookies政策廉正举报安全举报联系我们加入我们
阿里巴巴集团淘宝网天猫全球速卖通阿里巴巴国际交易市场1688阿里妈妈飞猪阿里云计算AliOS万网高德UC友盟优酷钉钉支付宝达摩院淘宝海外阿里云盘饿了么
© 2009-2023 Aliyun.com 版权所有 增值电信业务经营许可证： 浙B2-20080101 域名注册服务机构许可： 浙D3-20210002 京D3-20220015
浙公网安备 33010602009975号浙B2-20080101-4
关注我们：
新浪微博
联系我们
© 2009-2023 Aliyun.com 版权所有<br/>增值电信业务经营许可证： <a href="http://beian.miit.gov.cn/" target="_blank">浙B2-20080101</a> <br/>域名注册服务机构许可： <a href="https://domain.miit.gov.cn/域名注册服务机构/互联网域名/阿里云计算有限公司 " target="_blank">浙D3-20210002</a> <a href="https://domain.miit.gov.cn/域名注册服务机构/互联网域名/阿里巴巴云计算北京有限公司 " target="_blank">京D3-20220015</a>