从Spark导入

1clickhouse-spark-runtime-${spark_binary_version}_${scala_binary_version}-${version}.jar

1dependencies {
2  implementation("com.clickhouse.spark:clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}")
3  implementation("com.clickhouse:clickhouse-jdbc:{{ clickhouse_jdbc_version }}:all") { transitive = false }
4}

1repositries {
2  maven { url = "https://s01.oss.sonatype.org/content/repositories/snapshots" }
3}

1<dependency>
2  <groupId>com.clickhouse.spark</groupId>
3  <artifactId>clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}</artifactId>
4  <version>{{ stable_version }}</version>
5</dependency>
6<dependency>
7  <groupId>com.clickhouse</groupId>
8  <artifactId>clickhouse-jdbc</artifactId>
9  <classifier>all</classifier>
10  <version>{{ clickhouse_jdbc_version }}</version>
11  <exclusions>
12    <exclusion>
13      <groupId>*</groupId>
14      <artifactId>*</artifactId>
15    </exclusion>
16  </exclusions>
17</dependency>

1<repositories>
2  <repository>
3    <id>sonatype-oss-snapshots</id>
4    <name>Sonatype OSS Snapshots Repository</name>
5    <url>https://s01.oss.sonatype.org/content/repositories/snapshots</url>
6  </repository>
7</repositories>

1$SPARK_HOME/bin/spark-sql \
2  --conf spark.sql.catalog.clickhouse=com.clickhouse.spark.ClickHouseCatalog \
3  --conf spark.sql.catalog.clickhouse.host=${CLICKHOUSE_HOST:-127.0.0.1} \
4  --conf spark.sql.catalog.clickhouse.protocol=http \
5  --conf spark.sql.catalog.clickhouse.http_port=${CLICKHOUSE_HTTP_PORT:-8123} \
6  --conf spark.sql.catalog.clickhouse.user=${CLICKHOUSE_USER:-default} \
7  --conf spark.sql.catalog.clickhouse.password=${CLICKHOUSE_PASSWORD:-} \
8  --conf spark.sql.catalog.clickhouse.database=default \
9  --jars /path/clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}.jar,/path/clickhouse-jdbc-{{ clickhouse_jdbc_version }}-all.jar

1  --jars /path/clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}.jar,/path/clickhouse-jdbc-{{ clickhouse_jdbc_version }}-all.jar

1  --repositories https://{maven-cental-mirror or private-nexus-repo} \
2  --packages com.clickhouse.spark:clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }},com.clickhouse:clickhouse-jdbc:{{ clickhouse_jdbc_version }}:all

1spark-sql> use clickhouse;
2Time taken: 0.016 seconds
3
4spark-sql> create database if not exists test_db;
5Time taken: 0.022 seconds
6
7spark-sql> show databases;
8default
9system
10test_db
11Time taken: 0.289 seconds, Fetched 3 row(s)
12
13spark-sql> CREATE TABLE test_db.tbl_sql (
14         >   create_time TIMESTAMP NOT NULL,
15         >   m           INT       NOT NULL COMMENT 'part key',
16         >   id          BIGINT    NOT NULL COMMENT 'sort key',
17         >   value       STRING
18         > ) USING ClickHouse
19         > PARTITIONED BY (m)
20         > TBLPROPERTIES (
21         >   engine = 'MergeTree()',
22         >   order_by = 'id',
23         >   settings.index_granularity = 8192
24         > );
25Time taken: 0.242 seconds
26
27spark-sql> insert into test_db.tbl_sql values
28         > (timestamp'2021-01-01 10:10:10', 1, 1L, '1'),
29         > (timestamp'2022-02-02 10:10:10', 2, 2L, '2')
30         > as tabl(create_time, m, id, value);
31Time taken: 0.276 seconds
32
33spark-sql> select * from test_db.tbl_sql;
342021-01-01 10:10:10 1   1   1
352022-02-02 10:10:10 2   2   2
36Time taken: 0.116 seconds, Fetched 2 row(s)
37
38spark-sql> insert into test_db.tbl_sql select * from test_db.tbl_sql;
39Time taken: 1.028 seconds
40
41spark-sql> insert into test_db.tbl_sql select * from test_db.tbl_sql;
42Time taken: 0.462 seconds
43
44spark-sql> select count(*) from test_db.tbl_sql;
456
46Time taken: 1.421 seconds, Fetched 1 row(s)
47
48spark-sql> select * from test_db.tbl_sql;
492021-01-01 10:10:10 1   1   1
502021-01-01 10:10:10 1   1   1
512021-01-01 10:10:10 1   1   1
522022-02-02 10:10:10 2   2   2
532022-02-02 10:10:10 2   2   2
542022-02-02 10:10:10 2   2   2
55Time taken: 0.123 seconds, Fetched 6 row(s)
56
57spark-sql> delete from test_db.tbl_sql where id = 1;
58Time taken: 0.129 seconds
59
60spark-sql> select * from test_db.tbl_sql;
612022-02-02 10:10:10 2   2   2
622022-02-02 10:10:10 2   2   2
632022-02-02 10:10:10 2   2   2
64Time taken: 0.101 seconds, Fetched 3 row(s)

1$SPARK_HOME/bin/spark-shell \
2  --conf spark.sql.catalog.clickhouse=com.clickhouse.spark.ClickHouseCatalog \
3  --conf spark.sql.catalog.clickhouse.host=${CLICKHOUSE_HOST:-127.0.0.1} \
4  --conf spark.sql.catalog.clickhouse.protocol=http \
5  --conf spark.sql.catalog.clickhouse.http_port=${CLICKHOUSE_HTTP_PORT:-8123} \
6  --conf spark.sql.catalog.clickhouse.user=${CLICKHOUSE_USER:-default} \
7  --conf spark.sql.catalog.clickhouse.password=${CLICKHOUSE_PASSWORD:-} \
8  --conf spark.sql.catalog.clickhouse.database=default \
9  --jars /path/clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}.jar,/path/clickhouse-jdbc-{{ clickhouse_jdbc_version }}-all.jar

1  --jars /path/clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }}.jar,/path/clickhouse-jdbc-{{ clickhouse_jdbc_version }}-all.jar

1  --repositories https://{maven-cental-mirror or private-nexus-repo} \
2  --packages com.clickhouse.spark:clickhouse-spark-runtime-{{ spark_binary_version }}_{{ scala_binary_version }}:{{ stable_version }},com.clickhouse:clickhouse-jdbc:{{ clickhouse_jdbc_version }}:all

1scala> spark.sql("use clickhouse")
2res0: org.apache.spark.sql.DataFrame = []
3
4scala> spark.sql("create database test_db")
5res1: org.apache.spark.sql.DataFrame = []
6
7scala> spark.sql("show databases").show
8+---------+
9|namespace|
10+---------+
11|  default|
12|   system|
13|  test_db|
14+---------+
15
16scala> spark.sql("""
17     | CREATE TABLE test_db.tbl (
18     |   create_time TIMESTAMP NOT NULL,
19     |   m           INT       NOT NULL COMMENT 'part key',
20     |   id          BIGINT    NOT NULL COMMENT 'sort key',
21     |   value       STRING
22     | ) USING ClickHouse
23     | PARTITIONED BY (m)
24     | TBLPROPERTIES (
25     |   engine = 'MergeTree()',
26     |   order_by = 'id',
27     |   settings.index_granularity = 8192
28     | )
29     | """)
30res2: org.apache.spark.sql.DataFrame = []
31
32scala> :paste
33// Entering paste mode (ctrl-D to finish)
34
35spark.createDataFrame(Seq(
36    ("2021-01-01 10:10:10", 1L, "1"),
37    ("2022-02-02 10:10:10", 2L, "2")
38)).toDF("create_time", "id", "value")
39    .withColumn("create_time", to_timestamp($"create_time"))
40    .withColumn("m", month($"create_time"))
41    .select($"create_time", $"m", $"id", $"value")
42    .writeTo("test_db.tbl")
43    .append
44
45// Exiting paste mode, now interpreting.
46
47scala> spark.table("test_db.tbl").show
48+-------------------+---+---+-----+
49|        create_time|  m| id|value|
50+-------------------+---+---+-----+
51|2021-01-01 10:10:10|  1|  1|    1|
52|2022-02-02 10:10:10|  2|  2|    2|
53+-------------------+---+---+-----+
54
55scala> spark.sql("DELETE FROM test_db.tbl WHERE id=1")
56res3: org.apache.spark.sql.DataFrame = []
57
58scala> spark.table("test_db.tbl").show
59+-------------------+---+---+-----+
60|        create_time|  m| id|value|
61+-------------------+---+---+-----+
62|2022-02-02 10:10:10|  2|  2|    2|
63+-------------------+---+---+-----+

1scala> val options = Map(
2     |     "host" -> "clickhouse",
3     |     "protocol" -> "http",
4     |     "http_port" -> "8123",
5     |     "user" -> "default",
6     |     "password" -> ""
7     | )
8
9scala> val sql = """
10     | |CREATE TABLE test_db.person (
11     | |  id    Int64,
12     | |  name  String,
13     | |  age Nullable(Int32)
14     | |)
15     | |ENGINE = MergeTree()
16     | |ORDER BY id
17     | """.stripMargin
18
19scala> spark.executeCommand("com.clickhouse.spark.ClickHouseCommandRunner", sql, options) 
20
21scala> spark.sql("show tables in clickhouse_s1r1.test_db").show
22+---------+---------+-----------+
23|namespace|tableName|isTemporary|
24+---------+---------+-----------+
25|  test_db|   person|      false|
26+---------+---------+-----------+
27
28scala> spark.table("clickhouse_s1r1.test_db.person").printSchema
29root
30 |-- id: long (nullable = false)
31 |-- name: string (nullable = false)
32 |-- age: integer (nullable = true)

1public static void main(String[] args) {
2        // Initialize Spark session
3        SparkSession spark = SparkSession.builder().appName("example").master("local").getOrCreate();
4
5        // JDBC connection details
6        String jdbcUrl = "jdbc:ch://localhost:8123/default";
7        Properties jdbcProperties = new Properties();
8        jdbcProperties.put("user", "default");
9        jdbcProperties.put("password", "123456");
10
11        // Load the table from ClickHouse
12        Dataset<Row> df = spark.read().jdbc(jdbcUrl, "example_table", jdbcProperties);
13
14        // Show the DataFrame
15        df.show();
16
17        // Stop the Spark session
18        spark.stop();
19    }

1    public static void main(String[] args) {
2        // Initialize Spark session
3        SparkSession spark = SparkSession.builder().appName("example").master("local").getOrCreate();
4
5        // JDBC connection details
6        String jdbcUrl = "jdbc:ch://localhost:8123/default";
7        Properties jdbcProperties = new Properties();
8        jdbcProperties.put("user", "default");
9        jdbcProperties.put("password", "******");
10        // Create a sample DataFrame
11        StructType schema = new StructType(new StructField[]{
12                DataTypes.createStructField("id", DataTypes.IntegerType, false),
13                DataTypes.createStructField("name", DataTypes.StringType, false)
14        });
15        
16        List<Row> rows = new ArrayList<Row>();
17        rows.add(RowFactory.create(1, "John"));
18        rows.add(RowFactory.create(2, "Doe"));
19
20        Dataset<Row> df = spark.createDataFrame(rows, schema);
21
22        df.write()
23                .mode(SaveMode.Append)
24                .jdbc(jdbcUrl, "my_table", jdbcProperties);
25        // Show the DataFrame
26        df.show();
27
28        // Stop the Spark session
29        spark.stop();
30    }