مثال على بيانات استيراد الدُفعات إلى Elasticsearch بناءً على دفعة الربيع

1. مقدمة

عندما يكون للنظام كمية كبيرة من البيانات التي يجب استيرادها من قاعدة البيانات ، يمكن أن يؤدي استخدام دفعة الربيع إلى تحسين كفاءة الاستيراد. يستخدم Spring Batch itemReader لقراءة صفحات البيانات ، ويكتب العنصر بيانات البيانات على دفعات. نظرًا لأن Spring Batch لا يوفر eutwriter و itemReader الخاص بـ Elastisearch ، في هذا المثال ، يتم استخدام ElasticseArdeMwriter المخصص (ElasticsearchItemReader) لاستيراد الدُفعات.

2. مثال

2.1 pom.xml

تستخدم هذه المقالة مزاح بيانات الربيع لتوصيل ES (يمكنك أيضًا استخدام Spring Data Elasticsearch لتوصيل ES) ، وإصدار ES هو 5.5.3

 <؟ XSI: schemalocation = "http://maven.apache.org/pom/4.0.0 <soph> 0.0.1-snapshot </version> <packaging> جرة </packaging> <name> es-etl </name> <description> مشروع تجريبي لـ Spring Boot </description> <parent> <roupiD> org.springframework.boot </rougiD> <REPERANTERPARPHAPT/> <!-LANEUP PARTER من المستودع-> </inpart> <ferties> <project.build.sourceencoding> utf-8 </project.build.sourceencoding> <Project.reporting.outputing> uTF-8 </proport.reporting.outppouting> </sperties> <ependencies> <redence> <roupiD> org.springframework.boot </rougeid> <intifactid> spring-boot-starter </stifactid> </rependency> <rependency> <roupiD> org.springframework.boot </groupid> </reperency> <reperency> <roupiD> org.postgresql </rougiD> <artifactId> postgresql </shintifactid> </premited> <speration> <roupiD> org.springframework.boo </groupid> <roughid> com.github.vanroy </rougiD> <StifactId> Spring-Boot-Starter-Data-Jest </artifactId> <الإصدار> 3.0.0.Release </version> </dependency> <sperency> <prowEd> <roupend> io.searchbox </grouped> <roupl> org.projectlombok </rougiD> <StifactId> lombok </stifactid> </reperence> <reperency> <roupiD> org.springframework.boot </groupid> </intifactid> spring-boot-starter-test </artifactid> <scope> <Groper> org.springframework.boot </rougiD> <StifactId> Spring-boot-maven-plugin </shintifactid> </sultcin> </plugins> </build> <spositories> <spository> <Url> https://repo.spring.io/snapshot </url> <tainsts> <تمكين> صواب </enabled> </snapshots> </ropository> <rropository> <id> spring-milestones </id> <ame> milestones </name> <spleshots> <تمكين> خطأ </enupper> </snapshots> </ropository> </ropository> </ropository> </ropository> </suncloyrepository> <spleshots> <tenabled> true </texuped> </snapshots> </pluginRepository> <replopinRepository> <id> spring-milestones </id> <name> معالم الربيع </name> <Url> https://repo.spring.io/milestone </url> <slatshots> </sultiinrepository> </sugioninrepository> </sucplitiveository> </sucplitiveitory> </pluginRepositories> </project>

2.2 فئة الكيان والمستودع

 package com.hfcsbc.esetl.domain ؛ استيراد lombok.data ؛ استيراد org.springframework.data.elasticsearch.annotations.document ؛ استيراد org.springframework.data.elasticsearch.annotations.field ؛ org.springframework.data.elasticsearch.annotations.fieldtype ؛ استيراد javax.persistence.entity ؛ استيراد javax.persistence.id ؛ استيراد javax.persistence.onetoOne ؛/** * إنشاء بواسطة pengchao في 2018/2/23 */@الوثيقة = " RefreshInterval = "-1")@entity@datapublic person {id private id id ؛ اسم السلسلة الخاصة ؛ onetoOne@field (type = fieldtype.nested) عنوان العنوان الخاص ؛}

 package com.hfcsbc.esetl.domain ؛ استيراد lombok.data ؛ استيراد javax.persistence.entity ؛ استيراد javax.persistence.id ؛/** * إنشاء من قبل Pengchao في 2018/2/23 */intity@datapublic addrate {id private id ؛ اسم السلسلة الخاصة ؛}

 package com.hfcsbc.esetl.repository.jpa ؛ استيراد com.hfcsbc.esetl.domain

 package com.hfcsbc.esetl.repository.es ؛ import com.hfcsbc.esetl.domain طويل> {}

2.3 تكوين ElasticsearchItemWriter

 package com.hfcsbc.esetl.itemwriter ؛ استيراد com.hfcsbc.esetl.repository.es.espersonRepository ؛ استيراد com.hfcsbc.esetl.domain.person org.springframework.batch.core.stepexecution ؛ استيراد org.springframework.batch.core.stepexecutionlistener ؛ استيراد org.springframework.batch.itemwriter ؛ استيراد java.util.list ؛ itemWriter <Person> ، itemWritElistener <Person> ، StepExecutionListener {private espersonRepository personRepository ؛ Public ElasticSearchItemWriter (espersonRepository personRepository) {this.personRepository = personRepository ؛ } Override public void postwrite (قائمة <؟ تمديد الشخص> العناصر) {} Override public void Quotwrite (قائمة <؟ تمديد الشخص> العناصر)} Afterstep (StepExecution StepExecution) {return null ؛ } Override public void write (قائمة <؟ تمديد الشخص> العناصر) يلقي الاستثناء {// طريقة saveall التي تنفذ class ustrigtionelasticsearchrepository calls elasticsearchoperations.bulkindex (Queries) ، وهو مؤشر دفعة من personrepository.saveall (العناصر) ؛ }}

2.4 تكوين ElasticsearchItemReader (هذا المثال غير مستخدم ، للرجوع إليه فقط)

 package com.hfcsbc.esetl.itemreader ؛ استيراد org.springframework.batch.item.data.abstractPaginatedDataiteMreader org.springframework.data.elasticsearch.core.query.searchquery ؛ استيراد java.util.iterator ؛/** * إنشاء من قبل Pengchao في 2018/2/24 */public class elasticsearchiteMreader <merson> يمتد recctedpaginatedDateAteMreader <firectionments {private. استعلام Searchquery النهائي الخاص ؛ الفصل النهائي الخاص <؟ يمتد الشخص> TargetType ؛ Public ElasticsearchItemReader (ElasticSearchOprations ElasticSearchOperations ، searchquery query ، class <؟ Extends Person> targettype) {this.elasticsearchOperations = ElasticsearchOperations ؛ this.query = استعلام ؛ this.targetType = targetType ؛ } Override محمية ITERATOR <PORING> DOPAGEREAD () {return (iterator <Person>) ElasticSearchOprations.queryForList (Query ، TargetType) .iterator () ؛ } Override public void بعد propertiesset () يلقي الاستثناء {}}

2.5 التكوين المطلوب لتكوين دفعة الربيع

 package com.hfcsbc.esetl.config ؛ استيراد com.hfcsbc.esetl.itemwriter.elasticsearchitemwriter ؛ استيراد com.hfcsbc.esetl.repository.es.spersonRepository ؛ استيراد com.hfcsbc.esetl.domain.person ؛ org.springframework.batch.core.step ؛ استيراد org.springframework.batch.core.configuration.annotation.enableBatchProcessing org.springframework.batch.core.configuration.annotation.stepbuilderfactory ؛ استيراد org.springframework.batch.core.launch.support.runidincrementer ؛ استيراد org.springframework.batch.core.core.repository.jobository ؛ org.springframework.batch.core.repository.support.jobrepositoryfactorybean ؛ استيراد org.springframework.batch.itemreader ؛ استيراد org.springframework.batch.itemwriter ؛ استيراد org.springframework.batch.item.database.jpapagingitemreader org.springframework.context.annotation.bean ؛ استيراد org.springframework.context.annotation. في 2018/2/23 */@configuration@enableBatchProcessingPublic Class BatchConfig {Autowired private EspersonRepository personrepository ؛ bean public itemReader <Person> orderItemReader (entityManagerFactory entityManagerFactory) {jPapagingItemReader <Person> reader = new JPapagingItemReader <Person> () ؛ String sqlquery = "SELECT * from person" ؛ جرب {jpanativeQueryProvider <Person> QueryProvider = new jpanativeQueryProvider <Person> () ؛ queryprovider.setsqlquery (sqlquery) ؛ QueryProvider.setentityClass (person.class) ؛ QueryProvider.afterPropertiesset () ؛ reader.setentityManagerFactory (entityManagerFactory) ؛ reader.setpagesize (10000) ؛ reader.setQueryProvider (QueryProvider) ؛ reader.afterpropertiesset () ؛ reader.setsavestate (صواب) ؛ } catch (استثناء e) {E.PrintStackTrace () ؛ } قارئ العودة ؛ } bean public ElasticSearchItemWriter itemWriter () {return new ElasticseArdeMwriter (personRepository) ؛ } @bean الخطوة العامة الخطوة (StepBuilderFactory Stepbuilderfactory ، itemReader itemReader ، itemWriter itemWriter) {return stepbuilderfactory .get ("Step1") .Chunk (10000) .Reader (itemReader) .Riter (itemWriter) .build () ؛ } @Bean Public Job Job (JobBuilderFactory JobbuilderFactory ، الخطوة) {return jobbuilderfactory .get ("ImportJob") .incrementer (new RunidIncrementer () } /*** عند تنفيذ دفعة الربيع ، سيتم إنشاء بعض الجداول التي تحتاجها. فيما يلي موقع لإنشاء الجدول: DataSource * Param DataSource * Param Manager * @Return */ bean public JobRepository JobRepository (DataSource DataSource ، PlatformTackActionManager Manager) JobRepositoryFactoryBean.setDatasource (DataSource) ؛ JobRepositoryFactoryBean.setTransactionManager (مدير) ؛ JobRepositoryFactoryBean.setDatabasEtype ("postgres") ؛ حاول {return jobrepositoryfactorybean.getObject () ؛ } catch (استثناء e) {E.PrintStackTrace () ؛ } إرجاع فارغ ؛ }}

2.6 قم بتكوين عنوان اتصال قاعدة البيانات و ES

 الربيع: redis: المضيف: 192.168.1.222 البيانات: Jest: URI: JDBC: postgresql: //192.168.1.222: 5433/person اسم المستخدم: HFCB كلمة المرور: HFCB DRIVER-CLASS-NAME: ORG.POSTGRESQL.DRIVER MAX-ACTICT

2.7 تكوين فئة الدخول

 package com.hfcsbc.esetl ؛ استيراد org.springframework.boot.springapplication ؛ استيراد org.springframework.boot.autoconfigure.springbootapplication org.springframework.boot.autoconfigure.data.elasticsearch.elasticsearchdataautoConfiguration ؛ استيراد org.springframework.data.elasticsearch.repository.config.enableAlasticsearchRepositories ؛ استيراد org.springframework.data.jpa.repository.config.enablejparepositories ؛ springBoOtApplication (exclude = {elasticsearchautoconfiguration.class ، elasticsearchDataautoConfiguration.class})@enableEvoseActor "com.hfcsbc.esetl.repository")@enablejParePositories (basePackages = "com.hfcsbc.esetl.repository.jpa") الفئة العامة esetlapplication {public static void main (string [] }}

ما سبق هو كل محتوى هذه المقالة. آمل أن يكون ذلك مفيدًا لتعلم الجميع وآمل أن يدعم الجميع wulin.com أكثر.