一、案例概述案例名称：智能文化内容分析系统技术架构：华为开发者空间+MaaS服务+Web应用应用场景：文化作品智能分析、内容生成、关系图谱可视化二、场景痛点分析在文化内容创作和分析领域存在以下痛点：人物关系分析依赖人工梳理，效率低下多语言歌词翻译和理解门槛高影视作品画面分析需要专业影视知识文化数据统计整理工作繁琐三、技术方案设计3.1架构设计text前端界面→华为开发者空间(容器)→MaaSAPI→DeepSeek大模型↓数据存储与分析3.2使用产品华为开发者空间：云开发环境部署Web应用MaaS服务：调用DeepSeek-V3进行内容分析云容器引擎：应用部署和运行环境四、开发实践过程4.1环境准备bash#在华为开发者空间创建容器环境gitclonehttps://gitcode.com/your-repo/cultural-analysis.gitcdcultural-analysisnpminstall4.2MaaS服务集成python#调用DeepSeekAPI进行内容分析importrequestsdefanalyze_cultural_content(text):headers={'Authorization':'BearerYOUR_MAAS_API_KEY','Content-Type':'application/json'}data={'model':'deepseek-v3','messages':[{'role':'user','content':f'分析以下文化内容：{text}'}]}response=requests.post('https://maas-api.huaweicloud.com/v1/chat/completions',headers=headers,json=data)returnresponse.json()4.3核心功能实现智能人物关系分析自动提取文本中的人物关系生成关系图谱数据可视化展示多语言歌词理解歌词翻译和情感分析文化背景解读意境可视化影视画面智能解读镜头语言分析色彩运用识别情感表达解析五、应用效果展示5.1功能亮点自动化人物关系梳理，效率提升80%智能歌词分析和翻译准确率90%+影视画面专业分析，降低学习门槛一体化文化内容管理平台5.2界面展示六、技术价值6.1创新点多模态文化分析：结合文本、图像、音视频分析智能关系挖掘：基于大模型的深度关系理解可视化呈现：复杂文化数据的直观展示6.2行业价值教育领域：文化课程教学辅助工具文化研究：学术研究的智能助手内容创作：创作灵感和分析工具七、总结展望本项目充分验证了华为云MaaS服务在文化科技领域的应用潜力。通过华为开发者空间的云原生环境和MaaS平台的大模型能力，成功构建了一个功能完善的智能文化分析系统。未来规划：集成更多文化数据库增加实时协作功能拓展到更多文化细分领域

直播时间：2025/8/12 15：00-16：30 直播嘉宾：贺行简-DTSE开发者技术专家吕晨-DTSE开发者技术专家 立即报名，参与直播！直播间可抽取华为耳机、华为定制雨伞、定制T恤哦~cid:link_0 直播链接：cid:link_1 直播简介：华为云师资培训直播，带您掌握产业级大数据课程体系与华为开发者空间实战能力，助力高校数字化转型！  

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。1、Flink 部署系列本部分介绍Flink的部署、配置相关基础内容。2、Flink基础系列本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。3、Flik Table API和SQL基础系列本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。4、Flik Table API和SQL提高与应用系列本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。5、Flink 监控系列本部分和实际的运维、监控工作相关。二、Flink 示例专栏Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。本文介绍了source、transformations和sink的基本用法，下一篇将介绍各自的自定义用法。一、source-transformations-sink示例1、maven依赖下文中所有示例都是用该maven依赖，除非有特殊说明的情况。<properties>        <encoding>UTF-8</encoding>        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>        <maven.compiler.source>1.8</maven.compiler.source>        <maven.compiler.target>1.8</maven.compiler.target>        <java.version>1.8</java.version>        <scala.version>2.12</scala.version>        <flink.version>1.12.0</flink.version>    </properties>    <dependencies>            <dependency>            <groupId>jdk.tools</groupId>            <artifactId>jdk.tools</artifactId>            <version>1.8</version>            <scope>system</scope>            <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-clients_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-scala_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-java</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-streaming-scala_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-streaming-java_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-api-java-bridge_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-planner-blink_2.12</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-table-common</artifactId>            <version>${flink.version}</version>        </dependency>        <!-- 日志 -->        <dependency>            <groupId>org.slf4j</groupId>            <artifactId>slf4j-log4j12</artifactId>            <version>1.7.7</version>            <scope>runtime</scope>        </dependency>        <dependency>            <groupId>log4j</groupId>            <artifactId>log4j</artifactId>            <version>1.2.17</version>            <scope>runtime</scope>        </dependency>        <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <version>1.18.2</version>            <scope>provided</scope>        </dependency>        <dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.1.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.1.4</version></dependency>    </dependencies>2、source示例1）、基于Collectionimport java.util.Arrays;import org.apache.flink.api.common.RuntimeExecutionMode;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/** * @author alanchan * */public class Source_Collection {    /**     * 一般用于学习测试时使用      * 1.env.fromElements(可变参数);      * 2.env.fromColletion(各种集合);     * 3.env.generateSequence(开始,结束);      * 4.env.fromSequence(开始,结束);     *      * @param args 基于集合     * @throws Exception     */    public static void main(String[] args) throws Exception {        // env        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);        // source        DataStream<String> ds1 = env.fromElements("i am alanchan", "i like flink");        DataStream<String> ds2 = env.fromCollection(Arrays.asList("i am alanchan", "i like flink"));        DataStream<Long> ds3 = env.generateSequence(1, 10);//已过期，使用fromSequence方法        DataStream<Long> ds4 = env.fromSequence(1, 10);        // transformation        // sink        ds1.print();        ds2.print();        ds3.print();        ds4.print();        // execute        env.execute();    }}//运行结果3> 310> 106> 61> 19> 94> 42> 27> 75> 514> 114> 214> 314> 414> 514> 614> 711> 811> 911> 108> 811> i am alanchan12> i like flink10> i like flink9> i am alanchan2）、基于文件//自己创建测试的文件，内容都会如实的展示了，故本示例不再提供import org.apache.flink.api.common.RuntimeExecutionMode;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/** * @author alanchan * */public class Source_File {    /**     * env.readTextFile(本地/HDFS文件/文件夹),压缩文件也可以     *      *      * @param args     * @throws Exception     */    public static void main(String[] args) throws Exception {        // env        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);        // source        DataStream<String> ds1 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.txt");        DataStream<String> ds2 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/input/distribute_cache_student");        DataStream<String> ds3 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.tar.gz");        DataStream<String> ds4 = env.readTextFile("hdfs://server2:8020///flinktest/wc-1688627439219");        // transformation        // sink        ds1.print();        ds2.print();        ds3.print();        ds4.print();        // execute        env.execute();    }}// 运行结果//单个文件8> i am alanchan12> i like flink1> and you ?//目录15> 1, "英文", 901> 2, "数学", 7011> and you ?8> i am alanchan9> i like flink3> 3, "英文", 8613> 1, "数学", 905> 1, "语文", 50//tar.gz12> i am alanchan12> i like flink12> and you ?//hdfs6> (hive,2)10> (flink,4)15> (hadoop,3)3）、基于socketimport org.apache.flink.api.common.RuntimeExecutionMode;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;/** * @author alanchan *         在192.168.10.42上使用nc -lk 9999 向指定端口发送数据 *         nc是netcat的简称，原本是用来设置路由器,我们可以利用它向某个端口发送数据  *         如果没有该命令可以下安装 yum install -y nc *          */public class Source_Socket {    /**     * @param args     * @throws Exception      */    public static void main(String[] args) throws Exception {        //env        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);        //source        DataStream<String> lines = env.socketTextStream("192.168.10.42", 9999);                //transformation                //sink        lines.print();        //execute        env.execute();    }}验证步骤：1、先启动该程序2、192.168.10.42上输入3、观察应用程序的控制台输出3、transformations示例对流数据中的单词进行统计，排除敏感词vx:alanchanchnimport org.apache.flink.api.common.RuntimeExecutionMode;import org.apache.flink.api.common.functions.FilterFunction;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.datastream.KeyedStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;/** * @author alanchan * */public class TransformationDemo {    /**     * @param args     * @throws Exception      */    public static void main(String[] args) throws Exception {        // env        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);        // source        DataStream<String> lines = env.socketTextStream("192.168.10.42", 9999);        // transformation        DataStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {            @Override            public void flatMap(String value, Collector<String> out) throws Exception {                String[] arr = value.split(",");                for (String word : arr) {                    out.collect(word);                }            }        });        DataStream<String> filted = words.filter(new FilterFunction<String>() {            @Override            public boolean filter(String value) throws Exception {                return !value.equals("vx:alanchanchn");// 如果是vx:alanchanchn则返回false表示过滤掉            }        });        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = filted                .map(new MapFunction<String, Tuple2<String, Integer>>() {                    @Override                    public Tuple2<String, Integer> map(String value) throws Exception {                        return Tuple2.of(value, 1);                    }                });        KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(t -> t.f0);        SingleOutputStreamOperator<Tuple2<String, Integer>> result = grouped.sum(1);        // sink        result.print();        // execute        env.execute();    }验证步骤1、启动ncnc -lk 99992、启动应用程序3、在192.168.10.42中输入测试数据，如下4、观察应用程序的控制台，截图如下4、sink示例import org.apache.flink.api.common.RuntimeExecutionMode;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/** * @author alanchan  */public class SinkDemo {    /**     * @param args     * @throws Exception     */    public static void main(String[] args) throws Exception {        // env        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);        // source        DataStream<String> ds = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.txt");        System.setProperty("HADOOP_USER_NAME", "alanchan");        // transformation          // sink//        ds.print();//        ds.print("输出标识");        // 并行度与写出的文件个数有关，一个并行度写一个文件，多个并行度写多个文件        ds.writeAsText("hdfs://server2:8020///flinktest/words").setParallelism(1);        // execute        env.execute();    }}并行度为1的输出，到hdfs看结果————————————————                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY 版权协议，转载请附上原文出处链接和本声明。                        原文链接：https://blog.csdn.net/chenwewi520feng/article/details/131592673

华为云上搭建的高可用Flink集群最新版本是 flink1.18.0，jdk是21DLI山的最新flink只是 flink1.15，jdk每看到这两个有啥区别吗？DLI的flink版本为什么不和flink集群的保持一致呢？那我在本地基于flink1.18开发的程序，打包后是不是就不能提交到DLI上执行了？

客户目前的程序是基于flink1.13开发的，版本都是适配1.13的，如果迁移到DLI ，flink是1.15的,那么程序依赖connector是否能正常使用呢？

请问DLI Spark 作业支持读写华为云GeminiDB Cassandra 数据源吗？

为啥datatool的flink 写入表中会多出[""]

DLI解耦计算和存储负载为什么能提高性能？总所周知，网络传输数据延迟很大，远不如数据在单机上要来的快。那为啥分离存储和技术在不同主机，反而能提高性能呢

opcDA协议数据是否可以接入数据湖中？

1.资产轨迹Flink脚本中的时间加8小时逻辑描述2.为什么要加8小时，以致于轨迹时间不是当前时间，而是加8小时后的时间

需要将dws数据源A中的多张表join以后得结果集，写入另一个dws数据源中，有什么好办法？不想在源dws数据源中创建很多表，想将多表join后查询结果直接写入另一个dws数据源中

对于集成后在贴原层的数据，含有加密的数据，在有密钥的情况下如何进行解密，平台支持吗？求助大佬~

停止任务命令:flink stop d173e065c36b4385f2d386fb6e480274 -p obs://ddos/AIAE/savepoint/tianmen/ -yid application_1676631574356_0854报错截图 keeperErrorCode = Session closed because client failed to authenticate for /flink_base/flink zk截图

kafka消息正文-- key:`yd_test`.`test`, -- value:{ -- "mysqlType":{"name":"char","id":"int","age":"int"}, -- "id":606, -- "es":1662693580000, -- "ts":1662693580897, -- "database":"yd_test", -- "table":"test", -- "type":"INSERT", -- "isDdl":false, -- "sql":"", -- "sqlType":{"name":1,"id":4,"age":4}, -- "data":[{"name":"yd","id":"4","age":"95"}], -- "old":null, -- "pkNames":null -- }需要指定data的id字段为upsert-kafka源表的主键，怎么指定呢？经测试，我这样指定不对！--kafka源表 CREATE table test2(   data ARRAY>,   PRIMARY KEY (id) )    with (     "connector" = "upsert-kafka",      "properties.bootstrap.servers" = "",--kafka的内网连接地址和端口     "properties.group.id" = "",     "topic" = "",--Kafka的Topic     --'key.format' = 'json',     --'value.format' = 'json',     "scan.startup.mode" = "latest-offset" );

IDEA 上的database 连接数据湖