Разработка Flink — данные Mysql, импортированные в Hive
Разработка Flink — данные Mysql, импортированные в Hive

Предисловие

ResultSet в Mysql по умолчанию сохраняет результаты запроса в памяти. Если объем данных относительно велик, он будет занимать много памяти. Если памяти недостаточно, будет сообщено об ошибке.

Путь 1

Потоковая обработка результатов, позволяющая драйверу возвращать по одной строке данных за раз.

Язык кода:javascript
копировать
conn = DriverManager.getConnection("jdbc:mysql://192.168.7.102/", "user", "123456");
stmt = conn.createStatement(java.sql.ResultSet.TYPE_FORWARD_ONLY,java.sql.ResultSet.CONCUR_READ_ONLY);
stmt.setFetchSize(Integer.MIN_VALUE);
rs = stmt.executeQuery("SELECT * FROM tablename");

statementруководитьjava.sql.ResultSet.TYPE_FORWARD_ONLY,java.sql.ResultSet.CONCUR_READ_ONLYиstmt.setFetchSize(Integer.MIN_VALUE)комбинированные настройки,сообщит серверу MySQL о необходимости потоковой передачи возвращаемых результатов,Возвращать данные построчно.

Это настройка, требуемая mysql. Сначала я не понял, почему setFetchSize будет Integer.MIN_VALUE. Просто установите это обязательная настройка.

Примечание. При использовании этого метода для обработки данных необходимо обработать все данные в наборе результатов или закрыть набор результатов, прежде чем вы сможете использовать это соединение для следующего запроса и других операций, в противном случае будет выдано исключение.

Способ 2

Используя обработку на основе курсора, setFetchSize

Язык кода:javascript
копировать
conn = DriverManager.getConnection("jdbc:mysql://192.168.7.102/?useCursorFetch=true", "user", "123456");
stmt = conn.createStatement();
stmt.setFetchSize(100);
rs = stmt.executeQuery("SELECT * FROM tablename");

Mysql=>Hive

полагаться

Язык кода:javascript
копировать
<!--JSONинструмент-->
<dependency>
  <groupId>com.alibaba.fastjson2</groupId>
  <artifactId>fastjson2</artifactId>
  <version>2.0.22</version>
</dependency>

<!--действоватьMysql-->
<dependency>
  <groupId>mysql</groupId>
  <artifactId>mysql-connector-java</artifactId>
  <version>5.1.38</version>
</dependency>

<!--Hive JDBC-->
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-jdbc</artifactId>
  <version>2.1.0</version>
</dependency>

основной класс

Язык кода:javascript
копировать
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Mysql2Hive {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<JSONObject> mysqlData = env.addSource(new MysqlReader());
        mysqlData.addSink(new HiveWriter());
        mysqlData.print();
        env.execute("Mysql2Hive");
    }
}

Чтение MySQL

Язык кода:javascript
копировать
package com.xhkjedu.mysql2hive;

import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

class MysqlReader extends RichSourceFunction<JSONObject> {
    private transient Statement st = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://192.168.7.102/zdb?useCursorFetch=true", "root", "123456");
        st = con.createStatement();
        st.setFetchSize(3);
    }

    @Override
    public void run(SourceContext<JSONObject> ctx) throws Exception {

        ResultSet rs = st.executeQuery("select * from t_user01");

        while (rs.next()) {
            Integer id = rs.getInt("id");
            String name = rs.getString("name");

            JSONObject json = new JSONObject();
            json.put("rowKey", id);
            json.put("name", name);
            ctx.collect(json);
        }

        //rs.close();
        //st.close();
        //con.close();
    }

    @Override
    public void cancel() {

    }
}

ResultSet.next фактически принимает один фрагмент данных и связывается с базой данных, чтобы получить фрагмент данных. Он не извлекает все данные и не помещает их в память, поскольку перед ResultSet.next устанавливается соединение с базой данных. Соединение с базой данных отключено, вы не сможете получить данные. Объясните. Связь есть.

Напишите в Улей

Язык кода:javascript
копировать
package com.xhkjedu.mysql2hive;

import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;

class HiveWriter extends RichSinkFunction<JSONObject> {

    private transient Statement st = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:hive2://192.168.7.101:10000/default", "root", "123456");
        st = con.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void invoke(JSONObject json, Context context) throws Exception {
        Integer id = json.getInteger("id");
        String name = json.getString("name");
        String sql = "insert into t_user02(id,name) VALUES (" + id + ",'" + name + "')";
        System.out.println("Running: " + sql);
        st.execute(sql);
    }
}

Способ 2

основной класс

Язык кода:javascript
копировать
package com.xhkjedu.mysql2hive;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.types.Row;

public class Mysql2Hive {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Row> mysqlData = env.addSource(new MysqlReader());
        
        //Создаем улей catalog
        String name = "hive";      // Имя каталога, определяет уникальное представление имени.
        String defaultDatabase = "default";  // База по умолчанию данныхимя        String hiveConfDir = "/data/tools/bigdata/apache-hive-2.1.0-bin/conf";  // путь hive-site.xml

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
        tEnv.registerCatalog("myhive", hive);
        tEnv.useCatalog("myhive");
        tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        tEnv.useDatabase("default");
        Table table = tEnv.fromDataStream(mysqlData);
        table.execute().print();
//        tEnv.createTemporaryView("flink_user", table);
//        String insertSql = "insert into t_user02(id,name) SELECT id, name FROM flink_user";
//        tEnv.executeSql(insertSql);
    }

    private static RowTypeInfo getRowTypeInfo(Row row) {
        TypeInformation[] types = new TypeInformation[row.getArity()];
        String[] fieldNames = new String[row.getArity()];
        for (int i = 0; i < row.getArity(); i++) {
            Object field = row.getField(i);
            if (field instanceof Integer) {
                types[i] = BasicTypeInfo.INT_TYPE_INFO;
            } else {
                types[i] = BasicTypeInfo.STRING_TYPE_INFO;
            }
            fieldNames[i] = "f" + i;
        }
        return new RowTypeInfo(types, fieldNames);
    }
}

читать

Язык кода:javascript
копировать
package com.xhkjedu.mysql2hive;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

class MysqlReader extends RichSourceFunction<Row> {
    private transient Statement st = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://192.168.7.102/ztest?useCursorFetch=true", "root", "123456");
        st = con.createStatement();
        st.setFetchSize(3);
    }

    @Override
    public void run(SourceContext<Row> ctx) throws Exception {

        ResultSet rs = st.executeQuery("select * from t_user");

        while (rs.next()) {
            Integer id = rs.getInt("id");
            String name = rs.getString("name");

            ctx.collect(Row.ofKind(RowKind.INSERT, id, name));
        }

        //rs.close();
        //st.close();
        //con.close();
    }

    @Override
    public void cancel() {

    }
}

Поддерживаемые типы данных

В предыдущем примере DataStream, все типы данных в потоке определены POJO добрый. если DataStream Типы — это простые базовые типы. Могут ли они быть преобразованы непосредственно в таблицы? Это включает в себя таблицу середина Поддерживаемые типы данных。

В целом,DataStream середина Поддерживаемые типы данных,Table Оба поддерживаются, но при конвертации необходимо обратить внимание на некоторые детали.

1. Атомный тип

В Flink базовые типы данных (Целое, Двойное, Строковое) и общие типы данных (то есть типы данных, которые нельзя разделить снова) вместе называются «атомарными типами». Поток данных атомарного типа после преобразования становится таблицей только с одним столбцом. Тип данных поля столбца (поля) можно вывести из атомарного типа. Кроме того, вы также можете добавить параметры в метод fromDataStream() для переименования полей столбца.

Язык кода:javascript
копировать
// Воляпоток данныхпреобразован в Динамическая таблица, в динамической таблице только одно поле, переименуйте его в myLong
Table table = tableEnv.fromDataStream(stream, $(“myLong”));

2. Тип кортежа

Если атомарный тип не переименован, именем поля по умолчанию является «f0». Легко подумать, что на самом деле это результат обработки атомарного типа как кортежа Tuple1. Таблица поддерживает тип кортежа Tuple, определенный в Flink. Соответствующие имена полей в таблице по умолчанию являются именами атрибутов элементов в кортеже: f0, f1, f2... Все поля можно переупорядочить или извлечь подмножество полей. Поля также можно переименовать, вызвав метод выражения as().

Язык кода:javascript
копировать
// Воляпоток данныхпреобразован в содержит только f1 таблица полей
Table table = tableEnv.fromDataStream(stream, $("f1"));
// Воляпоток данныхпреобразован в содержит f0 и f1 таблица полей,существоватьповерхностьсередина f0 и f1 обмен позициями
Table table = tableEnv.fromDataStream(stream, $("f1"), $("f0"));
// Воля f1 Поля имеют названия myInt,f0 названный myLong
Table table = tableEnv.fromDataStream(stream, $("f1").as("myInt"), $("f0").as("myLong"));

3. Тип ПОЖО

Flink Он также поддерживает «составные типы», состоящие из нескольких типов данных. Наиболее типичный из них — простой. Java Объект(POJO тип). потому что POJO Имена полей с высокой надежностью были определены в этом виде преобразования потока данных. в Table Это казалось чрезвычайно гладким.

Преобразуйте поток данных типа POJO в таблицу. Если имя поля не указано, имя поля исходного типа POJO будет использоваться напрямую. Поля в POJO также можно переупорядочивать, извлекать и переименовывать, как показано в предыдущих примерах.

Язык кода:javascript
копировать
Table table = tableEnv.fromDataStream(stream);
Table table = tableEnv.fromDataStream(stream, $("user"));
Table table = tableEnv.fromDataStream(stream, $("user").as("myUser"), $("url").as("myUrl"));

4. Тип строки

Flink также определяет более распространенный тип данных в реляционных таблицах — строку, которая является основной организационной формой данных в таблице. Тип Row также является составным типом. Его длина фиксирована, и тип каждого поля не может быть напрямую выведен, поэтому при его использовании необходимо указать конкретную информацию о типе; оператор CREATE, который мы вызываем при создании таблицы, будет использовать все поля Name и; спецификация типа, которая называется «схемой» (Schema) таблицы во Flink. Кроме того, тип Row также имеет дополнительный атрибут RowKind, который используется для указания типа текущей строки в операции обновления. Таким образом, Row можно использовать для представления данных в потоке журнала изменений, тем самым обеспечивая преобразование между потоками и таблицами во Flink.

Таким образом, в потоке журнала обновлений тип элемента должен быть «Строка», и необходимо вызвать метод ofKind(), чтобы указать тип обновления. Вот конкретный пример:

Язык кода:javascript
копировать
DataStream<Row> dataStream = env.fromElements(
Row.ofKind(RowKind.INSERT, "Alice", 12),
Row.ofKind(RowKind.INSERT, "Bob", 5),
Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100));
// Поток журнала обновлений Воля преобразован в таблицу
Table table = tableEnv.fromChangelogStream(dataStream);
boy illustration
Неразрушающее увеличение изображений одним щелчком мыши, чтобы сделать их более четкими артефактами искусственного интеллекта, включая руководства по установке и использованию.
boy illustration
Копикодер: этот инструмент отлично работает с Cursor, Bolt и V0! Предоставьте более качественные подсказки для разработки интерфейса (создание навигационного веб-сайта с использованием искусственного интеллекта).
boy illustration
Новый бесплатный RooCline превосходит Cline v3.1? ! Быстрее, умнее и лучше вилка Cline! (Независимое программирование AI, порог 0)
boy illustration
Разработав более 10 проектов с помощью Cursor, я собрал 10 примеров и 60 подсказок.
boy illustration
Я потратил 72 часа на изучение курсорных агентов, и вот неоспоримые факты, которыми я должен поделиться!
boy illustration
Идеальная интеграция Cursor и DeepSeek API
boy illustration
DeepSeek V3 снижает затраты на обучение больших моделей
boy illustration
Артефакт, увеличивающий количество очков: на основе улучшения характеристик препятствия малым целям Yolov8 (SEAM, MultiSEAM).
boy illustration
DeepSeek V3 раскручивался уже три дня. Сегодня я попробовал самопровозглашенную модель «ChatGPT».
boy illustration
Open Devin — инженер-программист искусственного интеллекта с открытым исходным кодом, который меньше программирует и больше создает.
boy illustration
Эксклюзивное оригинальное улучшение YOLOv8: собственная разработка SPPF | SPPF сочетается с воспринимаемой большой сверткой ядра UniRepLK, а свертка с большим ядром + без расширения улучшает восприимчивое поле
boy illustration
Популярное и подробное объяснение DeepSeek-V3: от его появления до преимуществ и сравнения с GPT-4o.
boy illustration
9 основных словесных инструкций по доработке академических работ с помощью ChatGPT, эффективных и практичных, которые стоит собрать
boy illustration
Вызовите deepseek в vscode для реализации программирования с помощью искусственного интеллекта.
boy illustration
Познакомьтесь с принципами сверточных нейронных сетей (CNN) в одной статье (суперподробно)
boy illustration
50,3 тыс. звезд! Immich: автономное решение для резервного копирования фотографий и видео, которое экономит деньги и избавляет от беспокойства.
boy illustration
Cloud Native｜Практика: установка Dashbaord для K8s, графика неплохая
boy illustration
Краткий обзор статьи — использование синтетических данных при обучении больших моделей и оптимизации производительности
boy illustration
MiniPerplx: новая поисковая система искусственного интеллекта с открытым исходным кодом, спонсируемая xAI и Vercel.
boy illustration
Конструкция сервиса Synology Drive сочетает проникновение в интрасеть и синхронизацию папок заметок Obsidian в облаке.
boy illustration
Центр конфигурации————Накос
boy illustration
Начинаем с нуля при разработке в облаке Copilot: начать разработку с минимальным использованием кода стало проще
boy illustration
[Серия Docker] Docker создает мультиплатформенные образы: практика архитектуры Arm64
boy illustration
Обновление новых возможностей coze | Я использовал coze для создания апплета помощника по исправлению домашних заданий по математике
boy illustration
Советы по развертыванию Nginx: практическое создание статических веб-сайтов на облачных серверах
boy illustration
Feiniu fnos использует Docker для развертывания личного блокнота Notepad
boy illustration
Сверточная нейронная сеть VGG реализует классификацию изображений Cifar10 — практический опыт Pytorch
boy illustration
Начало работы с EdgeonePages — новым недорогим решением для хостинга веб-сайтов
boy illustration
[Зона легкого облачного игрового сервера] Управление игровыми архивами
boy illustration
Развертывание SpringCloud-проекта на базе Docker и Docker-Compose
©